操作场景
本文以调用 Java SDK 为例介绍通过开源 SDK 实现定时消息收发的操作过程。
前提条件
操作步骤
步骤1：安装 Java 依赖库
在 Java 项目中引入相关依赖，以 maven 工程为例，在 pom.xml 添加以下依赖：
说明
依赖版本要求 ≥ 4.9.3, 当前建议为4.9.4
<!-- in your <dependencies> block --><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.9.4</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-acl</artifactId><version>4.9.4</version></dependency>
步骤2：生产消息
创建消息生产者
// 实例化消息生产者ProducerDefaultMQProducer producer = new DefaultMQProducer(groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL权限);// 设置NameServer的地址producer.setNamesrvAddr(nameserver);// 启动Producer实例producer.start();
发送消息
固定延迟级别的消息
int totalMessagesToSend = 5;for (int i = 0; i < totalMessagesToSend; i++) {Message message = new Message(TOPIC_NAME, ("Hello scheduled message " + i).getBytes());// 设置消息延迟等级message.setDelayTimeLevel(5);// 发送消息SendResult sendResult = producer.send(message);System.out.println("sendResult = " + sendResult);}
任意延迟时间的消息
int totalMessagesToSend = 1;for (int i = 0; i < totalMessagesToSend; i++) {Message message = new Message(TOPIC_NAME, ("Hello timer message " + i).getBytes());// 设置发送消息的时间long timeStamp = System.currentTimeMillis() + 30000;// 若需要发送定时消息，则需要设置定时时间，消息将在指定时间进行投递，例如消息将在2022-08-08 08:08:08投递。// 若设置的时间戳在当前时间之前，则消息将被立即投递给Consumer。//将 __STARTDELIVERTIME 设定到 msg 的属性中message.putUserProperty("__STARTDELIVERTIME", String.valueOf(timeStamp));// 发送消息SendResult sendResult = producer.send(message);System.out.println("sendResult = " + sendResult);}
步骤3：消费消息
创建消费者
TDMQ RocketMQ 版支持 push 和 pull 两种消费模式。推荐Push消费模式
// 实例化消费者DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(groupName,new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL权限// 设置NameServer的地址pushConsumer.setNamesrvAddr(nameserver);
订阅消息
根据消费模式不同，订阅方式也有所区别。
// 订阅topicpushConsumer.subscribe(topic_name, "*");// 注册回调实现类来处理从broker拉取回来的消息pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {// 消息处理逻辑System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);// 标记该消息已经被成功消费, 根据消费情况，返回处理状态return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});// 启动消费者实例pushConsumer.start();
参数 | 说明 |
topic_name | 在控制台集群管理中 Topic 页签中复制具体 Topic 名称。 |
"*" | 订阅表达式如果为 null 或*表达式表示订阅全部，同时支持 "tag1 || tag2 || tag3" 标识订阅多个类型的 tag。 |
步骤4：查看消费详情
说明
