RocketMQ生产消息与消费消息超详细讲解

更新时间：2022年12月27日 13:48:04 作者：一个双子座的Java攻城狮

这篇文章主要介绍了RocketMQ生产消息与消费消息，RocketMQ可用于以三种方式发送消息：可靠的同步、可靠的异步和单向传输。前两种消息类型是可靠的，因为无论它们是否成功发送都有响应

1 RocketMQ简介

RocketMQ是阿里开源的一款非常优秀中间件产品，脱胎于阿里的另一款队列技术MetaQ，后捐赠给Apache基金会作为一款孵化技术，仅仅经历了一年多的时间就成为Apache基金会的顶级项目。并且它现在已经在阿里内部被广泛的应用，并且经受住了多次双十一的这种极致场景的压力（2017年的双十一，RocketMQ流转的消息量达到了万亿级，峰值TPS达到5600万）

2 MQ的常见产品

ActiveMQ：java语言实现，万级数据吞吐量，处理速度ms级，主从架构，成熟度高

RabbitMQ ：erlang语言实现，万级数据吞吐量，处理速度us级，主从架构，

RocketMQ ：java语言实现，十万级数据吞吐量，处理速度ms级，分布式架构，功能强，扩展性强

kafka ：scala语言实现，十万级数据吞吐量，处理速度ms级，分布式架构，功能较少，应用于大数据较多

3 环境搭建

创建maven工程

引入依赖：

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

4 单生产者单消费者模式

生产者：

//生产者，产生消息
public class Producer {
    public static void main(String[] args) throws Exception{
        //1.创建一个发送消息的对象Producer
        DefaultMQProducer producer=new DefaultMQProducer("group1");
        //2.设定发送的命名服务器地址
        producer.setNamesrvAddr("192.168.23.127:9876");
        //3启动发送的服务
        producer.start();
        //4.1创建要发送的消息对象,指定topic，指定内容body
        Message msg=new Message("topic1","hello rocketmq".getBytes("UTF-8"));
        //4.2发送消息
        SendResult result = producer.send(msg);
        System.out.println("返回结果:"+result);
        //5.关闭连接
        producer.shutdown();
    }
}

消费者：

//消费者，消费消息
public class Consumer {
    public static void main(String[] args) throws Exception{
        //1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("192.168.23.127:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1","*");
        //4.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    //System.out.println("收到消息："+msg);
                    System.out.println("消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功处理, mq 收到这个 标记后相同的消息讲不会再次发给消费者
            }
        });
        //5.启动接收消息的服务
        consumer.start();// 开启多线程 监控消息,持续运行
        System.out.println("接收消息服务已运行");
    }
}

测试：

5 单生产者多消费者模式

5.1默认模式（负载均衡）

生产者：

//生产者，产生消息
public class Producer {
    public static void main(String[] args) throws Exception{
        //1.创建一个发送消息的对象Producer
        DefaultMQProducer producer=new DefaultMQProducer("group1");
        //2.设定发送的命名服务器地址
        producer.setNamesrvAddr("192.168.23.127:9876");
        //3启动发送的服务
        producer.start();
        for (int i = 1; i <= 10; i++) {
            Message msg = new Message("topic1",("生产者2： hello rocketmq "+i).getBytes("UTF-8"));
            SendResult result = producer.send(msg);
            System.out.println("返回结果："+result);
        }
        //5.关闭连接
        producer.shutdown();
    }
}

消费者：

//消费者，消费消息
public class Consumer {
    public static void main(String[] args) throws Exception{
        //1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("192.168.23.127:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1","*");
        //4.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    //System.out.println("收到消息："+msg);
                    System.out.println("消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功处理, mq 收到这个 标记后相同的消息讲不会再次发给消费者
            }
        });
        //5.启动接收消息的服务
        consumer.start();// 开启多线程 监控消息,持续运行
        System.out.println("接收消息服务已运行");
    }
}

测试：

5.2广播模式

生产者的代码不变，消费者的代码改动如下：

		//设置当前消费者的消费模式（默认模式：负载均衡）
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置当前消费者的消费模式为广播模式：所有客户端接收的消息是一样的
        consumer.setMessageModel(MessageModel.BROADCASTING);

具体消费者代码：

//消费者，消费消息
public class Consumer {
    public static void main(String[] args) throws Exception{
        //1.创建一个接收消息的对象Consumer
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("group1");
        //2.设定接收的命名服务器地址
        consumer.setNamesrvAddr("192.168.23.127:9876");
        //3.设置接收消息对应的topic,对应的sub标签为任意*
        consumer.subscribe("topic1","*");
        //设置当前消费者的消费模式（默认模式：负载均衡）
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置当前消费者的消费模式为广播模式：所有客户端接收的消息是一样的
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.开启监听，用于接收消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    //System.out.println("收到消息："+msg);
                    System.out.println("消息："+new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功处理, mq 收到这个 标记后相同的消息讲不会再次发给消费者
            }
        });
        //5.启动接收消息的服务
        consumer.start();// 开启多线程 监控消息,持续运行
        System.out.println("接收消息服务已运行");
    }
}

测试：

广播模式的现象

如果生产者先发送消息, 后启动消费者, 消息只能被消费一次

如果多个消费者先启动(广播模式),后发消息,才有广播的效果

结论: 必须先启动消费者再启动发送者才有广播的效果

6 多生产者多消费者模式

多生产者产生的消息可以被同一个消费者消费，也可以被多个消费者消费

运行多个生产者，在启动消费者

测试：

到此这篇关于RocketMQ生产消息与消费消息超详细讲解的文章就介绍到这了,更多相关RocketMQ生产消息与消费消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

使用Hibernate根据实体类自动生成表的方法
这篇文章主要介绍了使用Hibernate根据实体类自动生成表的方法,该篇提供了两种方法,可以根据需要选择其一,希望对你有所帮助,如有不对的地方还望指正
2023-03-03
redis 获取 list 中的所有元素操作
这篇文章主要介绍了redis 获取 list 中的所有元素操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2021-03-03
Java锁的升级策略偏向锁轻量级锁重量级锁
在本文中小编给的大家整理了关于Java锁的升级策略偏向锁轻量级锁重量级锁的相关知识点内容，需要的朋友们参考下。
2019-06-06
SpringBoot启动执行sql脚本的3种方法实例
在应用程序启动后,可以自动执行建库、建表等SQL脚本,下面这篇文章主要给大家介绍了关于SpringBoot启动执行sql脚本的3种方法,文中通过实例代码介绍的非常详细,需要的朋友可以参考下
2022-01-01
Springboot如何优雅地进行字段校验
这篇文章主要给大家介绍了关于Springboot如何优雅地进行字段校验的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2021-01-01
JAVA设计模式之访问者模式原理与用法详解
这篇文章主要介绍了JAVA设计模式之访问者模式,简单说明了访问者模式的原理,并结合实例分析了java访问者模式的定义与用法,需要的朋友可以参考下
2017-08-08
基于mybatis batch实现批量提交大量数据
这篇文章主要介绍了基于mybatis batch实现批量提交大量数据,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-05-05
Spring Security入门demo案例
Spring Security是一个高度自定义的安全框架，本文主要介绍了Spring Security入门，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-08-08
Java语言实现简单的酒店前台管理小功能(实例代码)
这篇文章主要介绍了Java语言实现简单的酒店前台管理小功能，本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-03-03
java实例方法被覆盖,静态方法被隐藏Explain(详解)
下面小编就为大家带来一篇java实例方法被覆盖,静态方法被隐藏Explain(详解)。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-05-05