【MQ】Eclipse向RocketMQ中发送和接收消息

亦凉 2022-06-15 12:37 258阅读 0赞

# 一、前言 #

在前几篇博客中简单的向大家介绍了一下搭建RocketMQ的环境，搭建的过程也是由套路可寻的。既然是mq，就避免不了消息的发送和接收。所以在这篇博客中小编就向大家介绍一下mq的发送和接收。

# 二、关于RocketMQ的一些术语 #

小编认为，只要是接触过MQ的朋友，基本不会对下面的术语陌生，其中，

*  Producer  
          消息生产者，负责产生消息，一般由业务系统负责产生消息。
 *  Consumer  
          消息消费者，负责消费消息，一般是后台系统负责异步消费。
 *  Push Consumer  
          Consumer 的一种，应用通常向 Consumer 对象注册一个 Listener 接口，一旦收到消息，Consumer 对象立刻回调 Listener 接口方法。
 *  Pull Consumer  
          Consumer 的一种，应用通常主劢调用 Consumer 的拉消息方法从 Broker 拉消息，主动权由应用控制。
 *  Producer Group  
          一类 Producer 的集合名称，返类 Producer 通常収送一类消息，且发送逻辑一致。
 *  Consumer Group  
          一类 Consumer 的集合名称，返类 Consumer 通常消费一类消息，且消费逻辑一致。
 *  Broker  
          消息中转角色，负责存储消息，转发消息，一般也称为 Server。在 JMS 规范中称为 Provider。

# 三 、RocketMQ的Message数据结构 #

![这里写图片描述][SouthEast]

这张图片是来自官方文档。

在message中可以包含上面的几个属性，可以通过get、set的方式进行赋值：

msg.getTopic();
    msg.setTopic("Ares");

# 四、操作 #

当我们有了已经可以正常运行的RocketMQ服务器后，我们就可以开始建立自己的项目，进行数据的操作了。

## 4.1 建立Maven项目 ##

![这里写图片描述][SouthEast 1]

在pom.xml文件中添加依赖：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
      <modelVersion>4.0.0</modelVersion>
      <groupId>groupId</groupId>
      <artifactId>RocketMQ</artifactId>
      <version>0.0.1-SNAPSHOT</version>
    
      <dependencies>
            <dependency>
                <groupId>com.alibaba.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>3.0.10</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.rocketmq</groupId>
                <artifactId>rocketmq-all</artifactId>
                <version>3.0.10</version>
                <type>pom</type>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-classic</artifactId>
                <version>1.1.1</version>
            </dependency>
            <dependency>
                <groupId>ch.qos.logback</groupId>
                <artifactId>logback-core</artifactId>
                <version>1.1.1</version>
            </dependency>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.10</version>
                <scope>test</scope>
            </dependency>
        </dependencies>
    </project>

## 4.2 Producer ##

建立Producer类，主要是负责发送消息：

public class Producer {
    
        public static void main(String[] args) throws MQClientException{
    
            DefaultMQProducer producer = new DefaultMQProducer("producerGroupName2");
            producer.setNamesrvAddr("192.168.137.4:9876");
            producer.start();
            try {
                {
                    //实例化一个消息
                    Message msg = new Message("testTopic","tagA","keyA",("Hello World tagA!").getBytes("UTF-8"));
                    for(int i=50;i>0;i--){
                        if(i%2==0){
                            SendResult sendResult = producer.send(msg);
                            Thread.sleep(1000);
                            System.out.println("tagA send result:"+sendResult);
                        }else{
                            msg = new Message("testTopic","tagB","keyB",("Hello World tagB!").getBytes("UTF-8"));
                            SendResult sendResult = producer.send(msg);
                            Thread.sleep(1000);
                            System.out.println("tagB send result:"+sendResult);
                        }
                    }
                }
            }
            catch (Exception e) {
                e.printStackTrace();
            }
    
            Runtime.getRuntime().addShutdownHook(new Thread(() -> producer.shutdown()));
            System.exit(0);
        }
    }

注：  
      在这里我们通过“Message msg = new Message(“testTopic”,”tagA”,”keyA”,(“Hello World tagA!”).getBytes(“UTF-8”));”来实例化一个message，这个这里面的参数就如在第三点中介绍的那样。

## 4.3 consumer ##

建立Consumer类，主要负责发送消息：

public class Consumer {
    
        public static void main(String[] args) throws InterruptedException,
                MQClientException {
    
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                    "producerGroupName2");
            consumer.setNamesrvAddr("192.168.137.5:9876");
    
            consumer.subscribe("testTopic", "tagA || tagB");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                public ConsumeConcurrentlyStatus consumeMessage(
                        List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    System.out.println(Thread.currentThread().getName()
                            + " Receive New Messages: " + msgs);
                    MessageExt msg = msgs.get(0);
                    if (msg.getTopic().equals("testTopic")) {
                        if (msg.getTags() != null && msg.getTags().equals("tagA")) {
                            // 获取消息体
                            String message = new String(msg.getBody());
                            System.out.println("receive tagA message:" + message);
                        } else if (msg.getTags() != null
                                && msg.getTags().equals("tagB")) {
                            // 获取消息体
                            String message = new String(msg.getBody());
                            System.out.println("receive tagB message:" + message);
                        }
    
                    }
                    // 成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.println("Consumer Started.");
        }
    
    }

## 4.4 附加 ##

发送的消息都要序列化为二进制的消息体，可以通过gson完成这个操作。而且消息体可以为对象，也可以为集合，也可以为字符串，很多类型都可以使用。都可以解析。

msgBody = gson.toJson(msg).getBytes("UTF-8");

# 五、小结 #

通过这次的学习，小编亲手把各种类型的消息体都发送到了mq，然后在获取都可以正常解析，所以可以看出mq的效率还是很高的，操作性也很不错，所以大家可以尽情使用。最重要的是，他的承载数量很大，可以持久化上亿条数据。

[SouthEast]: /images/20220615/de672e1b48f5495c85067be03fd18056.png
[SouthEast 1]: /images/20220615/f971f8e81b194f3f91269d3f587e6e1a.png