Producer发送普通消息的方式
1.同步发送消息
同步消息代表发送端发送消息到broker之后，等待消息发送结果后，再次发送消息
实现步骤
- 创建生产端，声明在哪个生产组
- 注册NameServer地址
- 构建Message实体，指定topic、tag、body
- 启动生产端
- 发送消息
@Test
public void syncSend() throws MQBrokerException, RemotingException, InterruptedException, MQClientException {// 1.创建生产端，声明在哪个生产组DefaultMQProducer producer = new DefaultMQProducer("test_group");// 2.注册NameServer地址producer.setNamesrvAddr(NAME_SERVER_ADDR);// 3.构建Message实体，指定topic、tag、bodyMessage message = new Message("test", "hello world".getBytes());// 4.启动生产端producer.start();// 5.发送消息SendResult sendResult = producer.send(message);System.out.println(sendResult.getSendStatus());
}
2.异步发送消息
异步消息代表发送端发送完消息后，会直接返回，但是可以注册一个回调函数，当broker将消息落盘后，回调这个回调函数
实现步骤
- 创建生产端，声明在哪个生产组
- 注册NameServer地址
- 构建Message实体，指定topic、tag、body
- 启动生产端
- 发送消息，并且实现SendCallback接口
注：这里必须等待异步返回，否则消费者无法消费成功
@Test
public void asyncSend() throws RemotingException, InterruptedException, MQClientException {DefaultMQProducer producer = new DefaultMQProducer("test_group");producer.setNamesrvAddr(NAME_SERVER_ADDR);Message message = new Message("test", "tag-a","hello world".getBytes());producer.start();CountDownLatch countDownLatch = new CountDownLatch(1);// 发送消息，并且实现SendCallback接口producer.send(message, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {countDownLatch.countDown();System.out.println("发送成功：" + sendResult.getSendStatus());}@Overridepublic void onException(Throwable e) {countDownLatch.countDown();System.out.println("发送失败：" + e);}});countDownLatch.await();
}
3、发送单向消息
发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短
实现步骤
- 创建生产端，声明在哪个生产组
- 注册NameServer地址
- 构建Message实体，指定topic、tag、body
- 启动生产端
- 发送单向消息
@Test
public void sendOneWay() throws RemotingException, InterruptedException, MQClientException {DefaultMQProducer producer = new DefaultMQProducer("test_group");producer.setNamesrvAddr(NAME_SERVER_ADDR);Message message = new Message("test","tag-a", "hello world".getBytes());producer.start();producer.sendOneway(message);
}
Producer发送批量消息
在对吞吐率有一定要求的情况下，Apache RocketMQ可以将一些消息聚成一批以后进行发送，可以增加吞吐率，并减少API和网络调用次数。
@Test
public void sendBatch() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER_ADDR);// 构造批量消息List<Message> list = new ArrayList<>();list.add(new Message(RocketMQConfig.TEST_TOPIC, "hello world0".getBytes(Charset.defaultCharset())));list.add(new Message(RocketMQConfig.TEST_TOPIC, "hello world1".getBytes(Charset.defaultCharset())));list.add(new Message(RocketMQConfig.TEST_TOPIC, "hello world2".getBytes(Charset.defaultCharset())));producer.start();// 发送批量消息producer.send(list);producer.shutdown();
}
**注：**需要注意的是批量消息的大小不能超过 1MiB（否则需要自行分割），其次同一批 batch 中 topic 必须相同。
Producer发送延迟消息
Producer想要发送延迟消息，只要设置Message的DelayTimeLevel属性大于0即可。
RocketMQ无法随意设置延迟消息的延迟时间，只能根据延迟级别进行
延迟级别和延迟时间的对应关系
| 延迟级别 | 延迟时间 | 延迟级别 | 延迟时间 |
|---|---|---|---|
| 1 | 1s | 10 | 6min |
| 2 | 5s | 11 | 7min |
| 3 | 10s | 12 | 8min |
| 4 | 30s | 13 | 9min |
| 5 | 1min | 14 | 10min |
| 6 | 2min | 15 | 20min |
| 7 | 3min | 16 | 30min |
| 8 | 4min | 17 | 1h |
| 9 | 5min | 18 | 2h |
@Test
public void sendDelay() throws Exception {DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");producer.setNamesrvAddr(RocketMQConfig.NAME_SERVER_ADDR);producer.start();Message message = new Message(RocketMQConfig.TEST_TOPIC, "hello world".getBytes(Charset.defaultCharset()));// 设置延迟级别message.setDelayTimeLevel(3);// 发送批量消息SendResult sendResult = producer.send(message);System.out.println(sendResult.getSendStatus());producer.shutdown();
}
延迟消息的原理
延迟消息并不会直接发送到指定的topic，而是发送到一个延迟消息对应的topic中
当延迟消息的时间到达后，在将消息发送到指定的topic中
延迟消息投递的流程
-
producer端设置消息delayLevel延迟级别，消息属性DELAY中存储了对应了延时级别
-
broker端收到消息后，判断延时消息延迟级别，如果大于0，则备份消息原始topic，queueId，并将消息topic改为延时消息队列特定topic(SCHEDULE_TOPIC)，queueId改为延时级别的delayLevel-1
-
mq服务端ScheduleMessageService中，为每一个延迟级别单独设置一个定时器，定时(每隔1秒)拉取对应延迟级别的消费队列
-
根据消费偏移量offset从commitLog中解析出对应消息
-
从消息tagsCode中解析出消息应当被投递的时间，与当前时间做比较，判断是否应该进行投递
-
若到达了投递时间，则构建一个新的消息，并从消息属性中恢复出原始的topic，queueId，并清除消息延迟属性，从新进行消息投递
