网站商城微信支付宝支付宝支付接口,html网页代码成品,例点估算网站开发项目工作量,网名设计在线生成器Producer发送普通消息的方式
1.同步发送消息
同步消息代表发送端发送消息到broker之后#xff0c;等待消息发送结果后#xff0c;再次发送消息
实现步骤
创建生产端#xff0c;声明在哪个生产组注册NameServer地址构建Message实体#xff0c;指定topic、tag、body启动…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);// 构造批量消息ListMessage 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无法随意设置延迟消息的延迟时间只能根据延迟级别进行
延迟级别和延迟时间的对应关系
延迟级别延迟时间延迟级别延迟时间11s106min25s117min310s128min430s139min51min1410min62min1520min73min1630min84min171h95min182h
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则备份消息原始topicqueueId并将消息topic改为延时消息队列特定topic(SCHEDULE_TOPIC)queueId改为延时级别的delayLevel-1 mq服务端ScheduleMessageService中为每一个延迟级别单独设置一个定时器定时(每隔1秒)拉取对应延迟级别的消费队列 根据消费偏移量offset从commitLog中解析出对应消息 从消息tagsCode中解析出消息应当被投递的时间与当前时间做比较判断是否应该进行投递 若到达了投递时间则构建一个新的消息并从消息属性中恢复出原始的topicqueueId并清除消息延迟属性从新进行消息投递
