RocketMQ极简入门-RocketMQ普通消息发送

简介: RocketMQ已经写了两章了,一章是RocketMQ认识和安装,一章是RocketMQ的工作流程和核心概念,本章我们开始使用RocketMQ来发送和接收消息。RocketMQ的消息种类非常多,比如:普通消息,顺序消息,延迟消息,批量发送,消息过滤等等。本篇文章来探讨一下 普通消息的发送

前言

RocketMQ已经写了两章了,一章是RocketMQ认识和安装,一章是RocketMQ的工作流程和核心概念,本章我们开始使用RocketMQ来发送和接收消息。RocketMQ的消息种类非常多,比如:普通消息,顺序消息,延迟消息,批量发送,消息过滤等等。本篇文章来探讨一下 普通消息的发送

文章目录

普通消息发送

普通消息这里介绍三种发送方式,同步发送,异步发送,单向发送。我们先导入需要的依赖,版本尽量和RocketMQ的安装版本一致。

<dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.8.0</version></dependency></dependencies>

【注意】请保持 RocketMQ和Name Server 是启动状态,见RocketMQ安装》`

同步发送

同步消息是发送者发送消息,需要等待结果的返回,才能继续发送第二条消息,这是一种阻塞式模型,虽然消息可靠性高,但是阻塞导致性能低下。API : SendResult result = producer.send(message); 发送者代码示例:

japublicclassProducer {
//演示消息同步发送publicstaticvoidmain(String[] args) throwsInterruptedException, RemotingException, MQClientException, MQBrokerException {
//生产者DefaultMQProducerproducer=newDefaultMQProducer("syn-producerGroup");
//设置name server地址producer.setNamesrvAddr("127.0.0.1:9876");
//设置队列数量为2,默认为4,根据情况设置producer.setDefaultTopicQueueNums(2); 
//启动producer.start();
//发16个消息for (inti=0 ; i<16 ; i++){
Messagemessage=newMessage();
//消息主题message.setTopic("syn-topic");
//消息标签message.setTags("sms");
//添加内容message.setBody((i+"我是消息").getBytes(CharsetUtil.UTF_8));
//执行发送SendResultresult=producer.send(message);
//打印结果System.out.println(result);
        }
producer.shutdown();
    } 
 }

解释一下其中的相关类

  • DefaultMQProducer :生产者组,需要指定一个组名
  • producer.setNamesrvAddr(“127.0.0.1:9876”) :Name Server 的地址,生产者通过它来找到Bocker中的Topic(Topic默认可以自动创建的)
  • producer.setDefaultTopicQueueNums(2) :设置Topic中的队列的数量,默认是4个
  • producer.start() :启动 ,如果不启动程序不起作用
  • new Message :消息的对象封装,通过它设置 topic 主题,tags 标签,body 消息内容
  • SendResult result = producer.send(message) : 同步发送消息,可以立马得到返回结果,成功或者失败
  • producer.shutdown() : 关闭生产者

下面是 SendResult的结构

SendResult [
sendStatus=SEND_OK,         
msgId=C0A8006516B018B4AAC270EF9D940000,
offsetMsgId=C0A8006500002A9F0000000000008E1C, 
messageQueue=MessageQueue [
topic=syn-topic, 
brokerName=LAPTOP-20VLGCRC, 
queueId=3    ], 
queueOffset=0]
  • SendStatus : 状态OK
  • msgId: 发送者生成的ID
  • OffsetMsgId : 由Broker生成的消息ID
  • MessageQueue :队列信息

自行观察MQ的可视化插件界面,应该可以看到发送者发送过去的消息了。

异步发送

异步消息是发送者发送消息,无需等待发送结果就可以再发送第二条消息,它是通过回调的方式来获取到消息的发送结果,消息可靠性高,性能也高。API : producer.send(message,SendCallback) 示例代码:

//. . .上面案例一样,部分代码省略. . .producer.send(
//创建消息对象newMessage("asyn-topic", "sms", "我是消息".getBytes(CharsetUtil.UTF_8)),
//添加发送回调newSendCallback() {
//发送成功结果处理@OverridepublicvoidonSuccess(SendResultsendResult) {
System.out.println(sendResult);
            }
//发送异常结果处理@OverridepublicvoidonException(Throwablethrowable) {
System.out.println("发送异常:"+throwable.getMessage());
            }
        }
);

对于异步发送而言,发送完消息不会马上返回结果,也无需等待结果返回就能继续发送第二条消息。它通过 producer.send(message,SendCallback) -> SendCallback 回调来接收发送的结果,回调中包括了:onSuccess 和 onException两个回调方法来表示成功和失败。

单向发送

这种方式指的是发送者发送消息后无需等待Broker的结果返回,Broker也不会返回结果,消息是单向的,该方式性能最高,但是消息可靠性低。API : producer.sendOneway(message) 示例代码:

//. . .上面案例一样,部分代码省略. . .Messagemessage=newMessage("oneway-topic", "sms", "我是消息".getBytes(CharsetUtil.UTF_8));
producer.sendOneway(message);

通过 :producer.sendOneway(message)来发送消息是没有返回结果的,也无需任何等待,性能是最高的,但是数据的安全性最低,所以对于一些可被丢失的消息,比如:操作日志等就可以使用这种模式了。

消费者案例

下面是消费者端的代码

publicclassConsumer {
publicstaticvoidmain(String[] args) throwsMQClientException {
//创建消费者DefaultMQPushConsumerdefaultMQPushConsumer=newDefaultMQPushConsumer("syn-consumerGroup");
//设置name server 地址defaultMQPushConsumer.setNamesrvAddr("127.0.0.1:9876");
//从开始位置消费defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//广播模式defaultMQPushConsumer.setMessageModel(MessageModel.BROADCASTING);
//订阅defaultMQPushConsumer.subscribe("syn-topic","sms");
//注册消息监听器defaultMQPushConsumer.registerMessageListener(newMessageListenerConcurrently() {
@OverridepublicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>list, ConsumeConcurrentlyContextconsumeConcurrentlyContext) {
list.forEach(message->{
System.out.println(newString(message.getBody(), CharsetUtil.UTF_8));
                });
returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
defaultMQPushConsumer.start();
    }
}

解释一下相关的类

  • DefaultMQPushConsumer : 消费者组,基于push模式
  • defaultMQPushConsumer.setNamesrvAddr(“127.0.0.1:9876”) : name server地址
  • defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET):从哪个位置开始消费,FIRST代表最前面
  • defaultMQPushConsumer.setMessageModel(MessageModel.BROADCASTING) : 消息的消费模式广播模式,默认是 MessageModel.CLUSTERING 集群 模式。
  • defaultMQPushConsumer.subscribe(“syn-topic”,“sms”) :订阅哪个Topic中的哪个Tags中的消息
  • defaultMQPushConsumer.registerMessageListener :注册消息监听并处理消息,通常支持MessageListenerConcurrently并发和MessageListenerOrderly顺序 两种监听器。当监听器监听到消息,通过回调监听器中的 consumeMessage 方法来传递和处理消息。
  1. List<MessageExt> list : 消息列表 MessageExt中包含了消息的Body,消息的storeSzie,queueId等信息
  2. ConsumeConcurrentlyContext : 消费者上下文
  3. ConsumeConcurrentlyStatus :消息应答(签收),包括 CONSUME_SUCCESS 消费成功和 RECONSUME_LATER 消费失败两种结果。
  • defaultMQPushConsumer.start() :启动消费者,这个代码要写在注册了监听器的后面。

总结

下面对三种发送方式做一个对比

  • 可靠性最高: 同步发送 > 异步发送 > 单向发送
  • 性能最高:单向发送 > 异步发送 > 同步发送

使用场景建议如下

  • 如果是比较重要的不可丢失的消息,且对时效性要去不高建议使用同步发送,如转账消息
  • 如果是不重要的可失败的消息,比如日志消息,建议使用单向发送
  • 如果对时效性要求比较高,且消息不能丢失,可以尝试使用异步发送
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
29天前
|
消息中间件 Java Kafka
RabbitMQ 入门
RabbitMQ 入门
|
3月前
|
消息中间件 存储 Java
分享一下rocketmq入门小知识
分享一下rocketmq入门小知识
48 0
分享一下rocketmq入门小知识
|
4月前
|
消息中间件 新零售 弹性计算
云消息队列 RabbitMQ 版入门训练营,解锁对比开源优势与零基础实战
欢迎加入「云消息队列 RabbitMQ 版入门训练营」。
160 15
|
3月前
|
网络协议 物联网 测试技术
App Inventor 2 MQTT拓展入门(保姆级教程)
本文演示的是App和一个测试客户端进行消息交互的案例,实际应用中,我们的测试客户端可以看着是任意的、支持MQTT协议的硬件,通过订阅及发布消息,联网硬件与我们的App进行双向数据通信,以实现万物互联的智能控制效果。
185 2
|
4月前
|
消息中间件
云消息队列RabbitMQ版入门训练营 打卡领好礼
云消息队列RabbitMQ版入门训练营 打卡领好礼
54 3
|
3月前
|
消息中间件
快来体验 消息队列RabbitMQ版入门训练营 打卡学习领好礼
快来体验 消息队列RabbitMQ版入门训练营 打卡学习领好礼
61 0
EMQ
|
6月前
|
安全 网络性能优化
MQTT 5.0 报文(Packets)入门指南
MQTT 控制报文是 MQTT 数据传输的最小单元。MQTT 客户端和服务端通过交换控制报文来完成它们的工作,比如订阅主题和发布消息。
EMQ
699 7
MQTT 5.0 报文(Packets)入门指南
|
5月前
|
消息中间件 存储 Kafka
01.RabbitMQ入门
01.RabbitMQ入门
56 0
|
5月前
|
消息中间件 存储 前端开发
RabbitMQ在Java中的完美实现:从入门到精通
本文由木头左介绍如何在Java项目中使用RabbitMQ。RabbitMQ是开源的AMQP实现,支持多种客户端,适合分布式系统中的消息传递。首先需安装Erlang和RabbitMQ,接着在Java项目中添加RabbitMQ客户端库依赖。通过创建连接工厂和连接,建立与RabbitMQ的通信,并展示了创建连接和通道的代码示例。
|
6月前
|
消息中间件 Docker 微服务
RabbitMQ入门指南(十一):延迟消息-延迟消息插件
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了DelayExchange插件、延迟消息插件实现延迟消息等内容。
933 0