全部产品
云市场

收发定时消息

更新时间:2020-03-13 18:33:02

本文将引导您如何使用 TCP 协议下的 Java SDK 进行定时消息的收发。

前置条件

需要确保您已完成以下操作:

背景信息

通过定时消息,在消息发送后,可以在当前时间点之后的某一个时间点,再投递到消费者进行消费,适用于对消息生产和消费有时间窗口要求,或者利用消息触发定时任务的场景。

关于定时消息的更多信息,参见 消息类型 > 定时和延时消息

说明:对于新用户,建议在正式收发消息前,阅读 Demo 工程 来了解搭建消息队列工程的具体步骤。

发送定时消息

具体的示例代码,请以 消息队列代码库 为准。

发送定时消息的示例代码如下。

  1. import java.util.Properties;
  2. import java.util.concurrent.TimeUnit;
  3. import com.alipay.sofa.sofamq.client.PropertyKeyConst;
  4. import io.openmessaging.api.Message;
  5. import io.openmessaging.api.MessagingAccessPoint;
  6. import io.openmessaging.api.OMS;
  7. import io.openmessaging.api.OMSBuiltinKeys;
  8. import io.openmessaging.api.Producer;
  9. import io.openmessaging.api.SendResult;
  10. public class DelayProducerTest {
  11. public static void main(String... args) {
  12. Properties credentials = new Properties();
  13. // AccessKeyId 阿里云身份验证,在阿里云服务器管理控制台创建
  14. credentials.setProperty(OMSBuiltinKeys.ACCESS_KEY, "$accessKey");
  15. // AccessKeySecret 阿里云身份验证,在阿里云服务器管理控制台创建
  16. credentials.setProperty(OMSBuiltinKeys.SECRET_KEY, "xxxxx");
  17. // 设置 TCP 接入域名,进入控制台的概览页面查看接入点配置
  18. MessagingAccessPoint accessPoint = OMS.builder().driver("sofamq").endpoint("$endpoint")
  19. .withCredentials(credentials).build();
  20. Properties properties = new Properties();
  21. // 设置用户实例,进入控制台的概览页面查看接入点配置
  22. properties.setProperty(PropertyKeyConst.INSTANCE_ID, "$instanceId");
  23. properties.setProperty(PropertyKeyConst.GROUP_ID, "YOUR_GROUP");
  24. Producer producer = accessPoint.createProducer(properties);
  25. producer.start();
  26. Message message = new Message("$topic", "YOUR_TAG", "hello world".getBytes());
  27. // 定时消息,单位毫秒(ms),在指定时间戳(当前时间之后)进行投递,例如 2020-03-13 18:27:00
  28. long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2020-03-13 18:27:00").getTime();
  29. message.setStartDeliverTime(timeStamp);
  30. SendResult sendResult = producer.send(message);
  31. System.out.println(sendResult);
  32. }
  33. }

订阅定时消息

定时消息的订阅方式与普通消息一致,详见 订阅消息