RocketMQ (四) 使用RocketMQ原生API收发消息

文章目录


创建项目

pom文件

新建 maven 项目或 module,添加 rocketmq-client 依赖。

<?xml version="1.0" encoding="UTF-8"?>
<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>cn.tedu</groupId>
    <artifactId>rocketmq-api</artifactId>
    <version>1.0-SNAPSHOT</version>

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

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-store</artifactId>
            <version>4.7.1</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

同步消息

RocketMQ (四) 使用RocketMQ原生API收发消息
同步消息发送要保证强一致性,发到master的消息向slave复制后,才会向生产者发送反馈信息。

这种可靠性同步地发送方式使用的比较广泛,比如:重要的消息通知,短信通知。

RocketMQ (四) 使用RocketMQ原生API收发消息

生产者

package m1;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Scanner;

public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //新建生产者实例
        DefaultMQProducer p = new DefaultMQProducer("producer1");
        //设置 name server 地址
        p.setNamesrvAddr("192.168.64.141:9876");
        //启动生产者 连接服务器
        p.start();

        //消息数据封装到 Message 对象
        //发送
        while(true){
            System.out.println("输入消息:");
            String s = new Scanner(System.in).nextLine();
            /**
             * Topic --- 一级分类
             * Tag --- 二级分类(可选)
             */
            //发送的目的地、发送给谁、发送的消息
            Message msg = new Message("Topic1", "TagA", s.getBytes());//Topic1是自己在服务器上创建的
            SendResult r = p.send(msg);
            System.out.println("发送的消息:"+r);
        }
    }
}

RocketMQ (四) 使用RocketMQ原生API收发消息
RocketMQ (四) 使用RocketMQ原生API收发消息
RocketMQ (四) 使用RocketMQ原生API收发消息

消费者

消费者的要点:

1.push 和 pull

消费者有两种模式:push 和 pull。

push 模式由服务器主动向消费者发送消息;pull 模式由消费者主动向服务器请求消息。

在消费者处理能力有限时,为了减轻消费者的压力,可以采用pull模式。多数情况下都采用 pull 模式。

2.NameServer

消费者需要向 NameServer 询问 Topic 的路由信息。

3.Topic

从指定的Topic接收消息。Topic相当于是一级分类。

4.Tag

Topic 相当于是一级分类,Tag 相当于是2级分类。

  • 多个 Tag 可以这样写: TagA || TagB || TagC
  • 不指定 Tag,或者说接收所有的 Tag,可以写星号:*
package m1;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //新建消费者实例
        DefaultMQPushConsumer c = new DefaultMQPushConsumer("Consumer1");

        //设置 name server
        c.setNamesrvAddr("192.168.64.141:9876");

        //订阅消息(从哪订阅消息)
        /**
         * 标签设置:
         *      TagA
         *      TagB || TagC || TagD
         *      *
         */
        c.subscribe("Topic1", "TagA");//从topic1订阅标签A的消息

        //消息监听器
        //Concurrently监听器会启动多个线程,可以并行的处理多条消息
        c.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt ext : msgs) {//快捷键:msgs.for
                    String s = new String(ext.getBody());
                    System.out.println("收到:"+s);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        
        //启动
        c.start();
    }
}

RocketMQ (四) 使用RocketMQ原生API收发消息

异步消息

RocketMQ (四) 使用RocketMQ原生API收发消息
master 收到消息后立即向生产者进行反馈。之后再以异步方式向 slave 复制消息。

异步消息通常用在对响应时间敏感的业务场景,即发送端不能容忍长时间地等待Broker的响应。

生产者

package demo2;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Scanner;

/*
异步发送消息

一条消息送出后, 不必暂停等待服务器针对这条消息的反馈, 而是可以立即发送后续消息.
使用监听器, 以异步的方式接收服务器的反馈
 */
public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
        DefaultMQProducer p = new DefaultMQProducer("producer-demo2");
        p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
        p.start();

        p.setRetryTimesWhenSendAsyncFailed(0);

        String topic = "Topic2";
        String tag = "TagA";
        String key = "Key-demo2";


        while (true) {
            System.out.print("输入消息,用逗号分隔多条消息: ");
            String[] a = new Scanner(System.in).nextLine().split(",");

            for (String s : a) {
                Message msg = new Message(topic, tag, key, s.getBytes());

                p.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("\n\n消息发送成功 : "+sendResult);
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        System.out.println("\n\n消息发送失败");
                    }
                });

                System.out.println("--------------------消息已送出-----------------------");
            }
        }
    }
}

消费者

package demo2;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
/*
与 demo1.Consumer 完全相同
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo2");
        c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");

        c.subscribe("Topic2", "TagA");

        c.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    System.out.println(new String(msg.getBody()) + " - " + msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        c.start();
        System.out.println("开始消费数据");
    }
}

单向消息

RocketMQ (四) 使用RocketMQ原生API收发消息
这种方式主要用在不特别关心发送结果的场景,例如日志发送。

生产者

package demo3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Scanner;

/*
单向消息

消息发出后, 服务器不会返回结果
 */
public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
        DefaultMQProducer p = new DefaultMQProducer("producer-demo3");
        p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
        p.start();

        String topic = "Topic3";
        String tag = "TagA";

        while (true) {
            System.out.print("输入消息,用逗号分隔多条消息: ");
            String[] a = new Scanner(System.in).nextLine().split(",");
            for (String s : a) {
                Message msg = new Message(topic, tag, s.getBytes());
                p.sendOneway(msg);
            }
            System.out.println("--------------------消息已送出-----------------------");
        }
    }
}

消费者

package demo3;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/*
与 demo1.Consumer 完全相同
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo2");
        c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");

        c.subscribe("Topic3", "TagA");

        c.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    System.out.println(new String(msg.getBody()) + " - " + msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        c.start();
        System.out.println("开始消费数据");
    }
}

顺序消息

RocketMQ (四) 使用RocketMQ原生API收发消息
上图演示了 Rocketmq 顺序消息的基本原理:

  • 同一组有序的消息序列,会被发送到同一个队列,按照 FIFO 的方式进行处理
  • 一个队列只允许一个消费者线程接收消息,这样就保证消息按顺序被接收

下面以订单为例:

一个订单的顺序流程是:创建、付款、推送、完成。订单号相同的消息会被先后发送到同一个队列中。消费时,从同一个队列接收同一个订单的消息。

生产者

package demo4;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;
import java.util.Scanner;
/*
以下消息, 相同id的消息按顺序发送到同一个队列,
消费时也从同一个队列按顺序消费
                                              topic

                                        =======================  queue1
                                        =======================  queue2
111,消息1  111,消息2  111,消息3   ------->=======================  queue3
                                        =======================  queue4
222,消息1  222,消息2  222,消息3   ------->=======================  queue5
                                        =======================  queue6
333,消息1  333,消息2  333,消息3   ------->=======================  queue7
                                        =======================  queue8
                                                    ......
 */
public class Producer {
    static String[] msgs = {
            "15103111039,创建",
                                "15103111065,创建",
            "15103111039,付款",
                                                    "15103117235,创建",
                                "15103111065,付款",
                                                    "15103117235,付款",
                                "15103111065,完成",
            "15103111039,推送",
                                                    "15103117235,完成",
            "15103111039,完成"
    };

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer p = new DefaultMQProducer("producer-demo4");
        p.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
        p.start();

        String topic = "Topic4";
        String tag = "TagA";

        for (String s : msgs) {
            System.out.println("按回车发送此消息: "+s);
            new Scanner(System.in).nextLine();

            Message msg = new Message(topic, tag, s.getBytes());

            String[] a = s.split(",");
            long orderId = Long.parseLong(a[0]);

            /*
            MessageQueueSelector用来选择发送的队列,
            这里用订单的id对队列数量取余来计算队列索引

            send(msg, queueSelector, obj)
            第三个参数会传递到queueSelector, 作为它的第三个参数
             */
            SendResult r = p.send(msg, new MessageQueueSelector() {
                /*
                三个参数的含义:
                queueList: 当前Topic中所有队列的列表
                message: 消息
                o: send()方法传入的orderId
                 */
                @Override
                public MessageQueue select(List<MessageQueue> queueList, Message message, Object o) {
                    Long orderId = (Long) o;
                    //订单id对队列数量取余, 相同订单id得到相同的队列索引
                    long index = orderId % queueList.size();
                    System.out.println("消息已发送到: "+queueList.get((int) index));
                    return queueList.get((int) index);
                }
            }, orderId);

            System.out.println(r+"\n\n");
        }
    }
}

消费者

package demo4;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer c = new DefaultMQPushConsumer("consumer-demo4");
        c.setNamesrvAddr("192.168.64.151:9876;192.168.64.152:9876");
        
        c.subscribe("Topic4", "*");

        c.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                String t = Thread.currentThread().getName();

                for (MessageExt msg : list) {
                    System.out.println(t+" - "+ msg.getQueueId() + " - " +new String(msg.getBody()));
                }

                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        c.start();
        System.out.println("开始消费数据");
    }
}

延时消息

消息发送到 Rocketmq 服务器后, 延迟一定时间再向消费者进行投递。

延时消息的使用场景:

比如电商里,提交了一个订单就可以发送一个延时消息,1h后去检查这个订单的状态,如果还是未付款就取消订单释放库存。

生产者发送消息时,对消息进行延时设置:

msg.setDelayTimeLevel(3);

其中3 代表级别而不是一个具体的时间值,级别和延时时长对应关系是在MessageStoreConfig 类种进行定义的:

this.messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

对应关系表:

级别 延时时长
1 1s
2 5s
3 10s
4 30s
5 1m
6 2m
7 3m
8 4m
9 5m
10 6m
11 7m
12 8m
13 9m
14 10m
15 20m
16 30m
17 1h
18 2h

生产者

package m1;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.Scanner;

public class Producer {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //新建生产者实例
        DefaultMQProducer p = new DefaultMQProducer("producer1");
        //设置 name server 地址
        p.setNamesrvAddr("192.168.64.141:9876");
        //启动生产者 连接服务器
        p.start();

        //消息数据封装到 Message 对象
        //发送
        while(true){
            System.out.println("输入消息:");
            String s = new Scanner(System.in).nextLine();
            /**
             * Topic --- 一级分类
             * Tag --- 二级分类(可选)
             */
            //发送的目的地、发送给谁、发送的消息
            Message msg = new Message("Topic1", "TagA", s.getBytes());//Topic1是自己在服务器上创建的
            if(Math.random() < 0.5){
                msg.setDelayTimeLevel(3);//代表10秒
                System.out.println("这条消息延时10秒");
            }
            SendResult r = p.send(msg);
            System.out.println("发送的消息:"+r);
        }
    }
}

RocketMQ (四) 使用RocketMQ原生API收发消息

消费者

package m1;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @Authod yuhe
 * Create 2021-10-29-16:40
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //新建消费者实例
        DefaultMQPushConsumer c = new DefaultMQPushConsumer("Consumer1");

        //设置 name server
        c.setNamesrvAddr("192.168.64.141:9876");

        //订阅消息(从哪订阅消息)
        /**
         * 标签设置:
         *      TagA
         *      TagB || TagC || TagD
         *      *
         */
        c.subscribe("Topic1", "TagA");//从topic1订阅标签A的消息

        //消息监听器
        //Concurrently监听器会启动多个线程,可以并行的处理多条消息
        c.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt ext : msgs) {//快捷键:msgs.for
                    String s = new String(ext.getBody());
                    System.out.println("收到:"+s);
                }
                //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                /**
                 * 在消息处理失败时,可以告诉服务器,稍后重新发送消息,重新消费。
                 * 如果多次处理失败,最多会重试 18 次(18个延时级别),重试的时间间隔会越来越长
                 */
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        //启动
        c.start();
    }
}

RocketMQ (四) 使用RocketMQ原生API收发消息

批量消息

批量发送消息能显著提高传递小消息的性能。限制是这些批量消息应该有相同的topic,相同的waitStoreMsgOK,而且不能是延时消息。此外,这一批消息的总大小不应超过4MB。

生产者


消费者


消息过滤

Tag 过滤

Tag 可以满足大多数消息过滤的需求。使用 Tag 过滤非常简单,例如:

consumer.subscribe("Topic1", "TagA || TagB || TagC");

对自定义属性过滤

生产者可以在消息中添加自定义的属性:

msg.putUserProperty("prop1", "1");
msg.putUserProperty("prop2", "2");

消费者接收数据时,可以根据属性来过滤消息:

consumer.subscribe("Topic7", MessageSelector.bySql("prop1=1 or prop2=2"));

可以看到,自定义属性的过滤语法是 Sql 语法,RocketMQ只定义了一些基本语法来支持这个特性,支持的 Sql 过滤语法如下:

  • 数值比较,比如:>,>=,<,<=,BETWEEN,=;
  • 字符比较,比如:=,<>,IN;
  • IS NULL 或者 IS NOT NULL;
  • 逻辑符号 AND,OR,NOT;

生产者


消费者


事务消息

RocketMQ 提供了可靠性消息,也叫事务消息。下面分析一下其原理。

事务消息的原理

RocketMQ (四) 使用RocketMQ原生API收发消息
RocketMQ (四) 使用RocketMQ原生API收发消息
下面来看 RocketMQ 的事务消息是如何来发送“可靠消息”的,只需要以下三步:

  1. 发送半消息(半消息不会发送给消费者)
  2. 执行本地事务
  3. 提交消息

RocketMQ (四) 使用RocketMQ原生API收发消息
RocketMQ (四) 使用RocketMQ原生API收发消息
RocketMQ (四) 使用RocketMQ原生API收发消息
完成事务消息发送后,消费者就可以以正常的方式来消费数据。

RocketMQ 的自动重发机制在绝大多数情况下,都可以保证消息被正确消费。

假如消息最终消费失败了,还可以由人工处理进行托底。RocketMQ (四) 使用RocketMQ原生API收发消息
上面分析的是正常情况下的执行流程。下面再来看两种错误情况:

  1. 事务执行失败时回滚消息
  2. 服务器无法得知消息状态时,需要主动回查消息状态

回滚:
RocketMQ (四) 使用RocketMQ原生API收发消息
消息回查:
RocketMQ (四) 使用RocketMQ原生API收发消息
RocketMQ (四) 使用RocketMQ原生API收发消息

生产者


消费者


上一篇:【架构师面试-消息队列-7】-常见消息队列产品使用场景与优缺点对比


下一篇:rocketmq传输协议