RocketMQ应用

这个名字先用着

2024-04-16 帮助1人

一、普通消息：

1.消息发送分类：

学新通

（1）同步发送消息：

学新通

（2）异步发送消息：

学新通

（3）单向发送消息：

学新通

2.普通消息代码举例：

（1）创建maven工程: rocketmq-test

（2）导入依赖：

   <dependencies>
        <!--rocketmq-client需要与本地安装的版本相同-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.2</version>
        </dependency>
    </dependencies>

学新通

（3）准备工作：

然后单机启动rocketmq:
单机启动：

cd /usr/src/software/rocketmq-4.9.2
启动：nohup sh bin/mqnamesrv &
查看日志：tail -f ~/logs/rocketmqlogs/namesrv.log

cd /usr/src/software/rocketmq-4.9.2
nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log

关闭防火墙：systemctl stop firewalld

然后启动控制台：http://localhost:6789/#/

学新通

（4）代码：

定义生产者：

（1）同步发消息生产者：Sync

学新通

com.fan.general.SyncProducer :

package com.fan.general;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
//同步发行消息
public class SyncProducer {
    public static void main(String[] args) throws Exception{
        //创建一个生产者producer,参数为创建一个生产者producer Group 名称
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        //指定nameserver地址
        producer.setNamesrvAddr("rocketmqOS:9876");
        //设置当前发送失败后的重试次数，默认不设置是2次
        producer.setRetryTimesWhenSendFailed(3);
        //设置发送超时实现为5秒，默认是3秒
        producer.setSendMsgTimeout(5000);
        //开启生产者
        producer.start();
        //生产并发送100条消息
        for (int i = 0; i < 100; i  ) {
            byte[] body = ("Hi,"   i).getBytes();
            Message msg = new Message("someTopic", "someTag", body);
            //为消息指定key
            msg.setKeys("key-"   i );
            //发送消息
            SendResult sendResult = producer.send(msg);
            System.out.println(sendResult);
        }
        //关闭producer
        producer.shutdown();
    }
}

涉及的类：
学新通

学新通

学新通
测试：

学新通

在linux主机上查看：
学新通

学新通

查看发送来的消息：

学新通

（2）异步消息发送生产者：Async

package com.fan.general;

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;

public class AsyncProducer {
    public static void main(String[] args)throws  Exception {
        //创建一个生产者
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        //指定服务器地址
        producer.setNamesrvAddr("rocketmqOS:9876");
        //指定异步发送失败后，不进行重试发送,也可改成2
        producer.setRetryTimesWhenSendAsyncFailed(0);
        //指定新创建的topic的queue数量为2，默认为4
        producer.setDefaultTopicQueueNums(2);
        //启动生产者
        producer.start();
        //发送100条消息
        for (int i = 0; i < 100; i  ) {
            //创建一个消息体
            byte[] body = ("hi,"   i).getBytes();
            Message message = new Message("myTopicA", "myTagA", body);
            //异步发送需要有callback回调方法
            producer.send(message, new SendCallback() {
                //当producer接收到mq发送来的ack后就会触发该回调方法的执行
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    throwable.printStackTrace();
                }
            });

        }//end-for
        //sleep一会了，由于是异步发送，这里如果不sleep,则消息还未发送就会将producer给关闭了
        Thread.sleep(3000);
        producer.shutdown();
    }
}

学新通

运行测试：

学新通

控制台看看消息：

学新通

（3）单向消息发送生产者：

package com.fan.general;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class OnewayProducer {
    public static void main(String[] args)throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr("rocketmqOS:9876");
        //启动生产者
        producer.start();
        for (int i = 0; i < 100; i  ) {
            byte[] msgBody = ("hi,"   i).getBytes();
            Message message = new Message("single", "singleTag", msgBody);
            //单向发送，没有消息的返回值，所以也不用打印
            producer.sendOneway(message);
        }
        producer.shutdown();
        System.out.println("producer 关闭");
    }
}

学新通

定义消费者：

SomeConsumer:

package com.fan.general;


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.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class SomeConsumer {
    public static void main(String[] args)throws MQClientException {
        //定义一个pull消费者
        //DefaultLitePullConsumer consumer1 = new DefaultLitePullConsumer("cg");
        //定义一个push 消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
        consumer.setNamesrvAddr("rocketmqOS:9876");
        //从哪里开始消费,指定从第一条开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //指定消费topic与tag
        consumer.subscribe("someTopic","*");
        //指定费用广播模式  进行消费，默认为集群模式的
        //consumer.setMessageModel(MessageModel.BROADCASTING);
        //注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs ) {
                    System.out.println(msg);
                }
                //返回消费状态：消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //开启消费者进行消费
        consumer.start();
        System.out.println("消费者开始了----");
    }
}

lite:简化的

学新通

运行测试：网页反应比较慢

学新通

二、顺序消息：

学新通

代码实现：

package com.fan.general;

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 java.util.List;

public class OrderedProducer {
    public static void main(String[] args)throws Exception {
        //创建一个生产者
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr("rocketmqOS:9876");
        //若为全局有序，需要设置queue数量为1
       //producer.setDefaultTopicQueueNums(1);
        producer.start();
        for (int i = 0; i < 100; i  ) {
            Integer orderId = i;
            byte[] body = ("hi,"   i).getBytes();
            Message msg = new Message("TopicA", "TagA", body);
            //将orderid作为消息key
            msg.setKeys(orderId.toString());
            //send()的第三个参数会传递给选择器的select()的第三个参数 ，该send为同步发送
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                //具体的选择算法在该方法中定义
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    //以下是使用消息key作为选择的选择算法
                    String keys = msg.getKeys();
                    Integer id = Integer.valueOf(keys);
                    //以下是使用arg作为选择key的选择算法
                    //Integer id = (Integer)arg;
                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            },orderId);
            System.out.println(sendResult);
        }
        producer.shutdown();
    }
}

运行测试：启动mq,记得关闭防火墙：systemctl stop firewalld

学新通

三、延迟消息：

学新通

具体步骤：
学新通

学新通

生产者：

package com.fan.delay;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import java.text.SimpleDateFormat;
import java.util.Date;
//消息的生产者
public class DelayProducer {
    public static void main(String[] args)throws Exception {
        //1.创造生产者
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        //2.给生产者设置名称服务器
        producer.setNamesrvAddr("rocketmqOS:9876");
        //3.开启生产者
        producer.start();
        for (int i = 0; i < 1; i  ) {
            byte[] body = ("hi,"   i).getBytes();
            //4.循环生产消息
            Message message = new Message("TopicB", "TagB", body);
            //指定消息的延时等级为3级，即延迟10秒
            message.setDelayTimeLevel(3);
            SendResult sendResult = producer.send(message);
            //输出消息被发送的时间
            System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date() ));
            System.out.println("," sendResult);
        }
        //5.生产完消息后关闭生产者
        producer.shutdown();
    }
}

消费者：

package com.fan.delay;

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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
//消息的消费者
public class OtherConsumer {
    public static void main(String[] args)throws Exception {
        //0.创造消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
        //1.给消费者设置名称服务器
        consumer.setNamesrvAddr("rocketmqOS:9876");
        //2.订阅的主题和子表达式是什么
        consumer.subscribe("TopicB","*"); //subscribe订阅
        //3.设置从哪里开始消费消息
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //4.注册消息监听器,参数是并发消息监听器，Concurrently并发的意思
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) { //逐条消费消息
                    //输出消息的消费时间
                    System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    System.out.println("," msg);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.开启消费者
        consumer.start();
        System.out.println("消费者已开启消费---");
    }
}

rocketmq消费者注册监听有两种模式，有序消费MessageListenerOrderly和并发消费MessageListenerConcurrently，这两种模式返回值不同。

MessageListenerOrderly正确消费返回ConsumeOrderlyStatus.SUCCESS，

稍后消费返回ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT

MessageListenerConcurrently正确消费返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS

稍后消费返回ConsumeConcurrentlyStatus.RECONSUME_LATER

运行测试：先启动消费者，后启动生产者：

生产者39秒开始生产消息，消费者49秒的时候开始消费消息：
学新通

学新通

四、事务消息：

学新通

代码演示：

定义事务消息生产者：

package com.fan.transaction;

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.*;

public class TransactionProducer {
    public static void main(String[] args)throws Exception {
        TransactionMQProducer producer = new TransactionMQProducer("tpg");
        producer.setNamesrvAddr("rocketmqOS:9876");
        /**
         * 定义一个线程池
         * @param corePoolSize 线程池中核心线程数量
         * @param maximumPoolSize 线程池中最多线程数
         * @param keepAliveTime 这是一个时间。当线程池中线程数量大于核心线程数量时,多余空闲线程的存活时长
         * @param unit 时间单位
         * @param workQueue 临时存放任务的队列，其参数就是队列的长度,即ArrayBlockingQueue
         * @param threadFactory 线程工厂
         */
        ExecutorService executorService = new ThreadPoolExecutor(2,
                5,
                100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("client-transation-msg-check-thread");
                        return thread;
                    }
                });

        //为生产者指定一个线程池
        producer.setExecutorService(executorService);
        //为生产者添加事务监听器
        producer.setTransactionListener(new ICBCTransactionListener());
        //开启生产者
        producer.start();

        String[] tags = {"TAGA","TAGB","TAGC"};
        for (int i = 0; i < 3; i  ) {
            byte[] body = ("hi,"   i).getBytes();
            Message msg = new Message("someTopic", tags[i], body);
            //发送事务消息
            //第二个参数用于指定在执行本地事务时要使用的业务参数
            TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.println("发送结果为："   sendResult.getSendStatus());
        }
    }
}

定义工行事务监听器：

package com.fan.transaction;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

public class ICBCTransactionListener implements TransactionListener {
    //回调操作方法，消息预提交成功就会触发该方法的执行，用于完成本地事务
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("预提交消息成功：" msg);
        //假设收到的TAGA的消息表示扣款成功；TAGB的消息表示扣款失败；
        // TAGC表示扣款结果不清楚，需要执行消息会查
        if(StringUtils.equals("TAGA",msg.getTags())){
            return LocalTransactionState.COMMIT_MESSAGE;//本地事务状态：提交消息
        }else if(StringUtils.equals("TAGB",msg.getTags())){
            return LocalTransactionState.ROLLBACK_MESSAGE;//本地事务状态：回滚消息
        }else if(StringUtils.equals("TAGC",msg.getTags())){
            return LocalTransactionState.UNKNOW;//本地事务状态：未知
        }
        return LocalTransactionState.UNKNOW;
    }

    //消息回查方法
    //引发消息回查的原因有两个：1.回调操作返回UNKNOW，2.TC没有接收到TM的最终全局事务确认指令
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("执行消息回查" msg.getTags());
        return LocalTransactionState.COMMIT_MESSAGE;//回查后返回：提交消息
    }
}

运行测试：

学新通

定义消费者：

直接使用普通消息的作为消费者即可：

package com.fan.transaction;

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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class TransactionConsumer {
    public static void main(String[] args)throws Exception  {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
        consumer.setNamesrvAddr("rocketmqOS:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TTopic","*");

        //注册并发消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                //逐条消费消息
                for (MessageExt msg : msgs ) {
                    System.out.println(msg);
                }
                //返回消费状态：消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //开启消费者进行消费
        consumer.start();
    }
}

控制台查看消息：

学新通

五、批量消息：

1.批量发送消息：

学新通

2.批量消费消息：

学新通

3.代码举例：

学新通

学新通
消息生产者：

package com.fan.batch;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.ArrayList;
import java.util.List;

public class BatchProducer {
    public static void main(String[] args)throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr("rocketmqOS:9876");
        //指定要发送的消息的最大大小。默认是4M
        /*
            不过，仅修改该属性是 不行的，还需要同时修改broker加载的配置文件中的maxMessageSize属性
            producer.setMaxMessageSize(8 * 1024 * 1024);
        */
        producer.start();
        ArrayList<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100; i  ) {
            byte[] body = ("hi,"   i).getBytes();
            Message msg = new Message("BatchTopic", "BatchTag", body);
            //将每一条消息放到消息集合中
            messages.add(msg);
        }//end-for
        //定义消息列表分割器，将 消息列表分割为多个不超过4M大小的小列表
        MessageListSplitter splitter = new MessageListSplitter(messages);
        while(splitter.hasNext()){
            try {
                List<Message> listItem = splitter.next();
                producer.send(listItem);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}

消息分割器：

package com.fan.batch;
import org.apache.rocketmq.common.message.Message;
import java.util.*;
import java.util.function.Consumer;
/*
消息列表分割器：其只会处理每条消息的大小不超过4M的情况。
若存在某条消息，其本身大小大于4M，这个分割器无法处理，其
直接将这条消息构成一个子列表返回，并没有再进行分割。
*/
public class MessageListSplitter implements Iterable<List<Message>>{
    //指定极限值为4M
    private final int SIZE_LIMIT = 4 * 1024 * 1024;
    //存放所有要发送的消息
    private final List<Message> messages;
    //要进行批量发送消息的小集合  起始索引
    private int currIndex;
    public MessageListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    public boolean hasNext() {
        return currIndex < messages.size();
    }

    public List<Message> next() {
        int nextIndex = currIndex;
        //记录当前要发送的这一小批次消息的大小
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex  ) {
            //获取当前遍历的消息
            Message message = messages.get(nextIndex);
            //统计当前遍历的message的大小
            int tmpSize = message.getTopic().length()   message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize  = entry.getKey().length()   entry.getValue().length();
            }
            tmpSize = tmpSize   20;
            if(tmpSize > SIZE_LIMIT){
                if(nextIndex - currIndex == 0){
                    nextIndex  ;
                }
                break;
            }
            if(tmpSize   totalSize > SIZE_LIMIT){
                break;
            }else{
                totalSize  = tmpSize;
            }
        }//end-for
        //获取当前messages列表的子集合【currIndex,nextIndex)
        List<Message> subList = this.messages.subList(currIndex, nextIndex);
        //下次遍历的开始索引
        currIndex = nextIndex;
        return subList;
    }

    @Override
    public Iterator<List<Message>> iterator() {
        return null;
    }

    @Override
    public void forEach(Consumer<? super List<Message>> action) {

    }

    @Override
    public Spliterator<List<Message>> spliterator() {
        return null;
    }
}

学新通

消费者：

package com.fan.batch;

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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class BatchConsumer {
    public static void main(String[] args)throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
        consumer.setNamesrvAddr("rocketmqOS:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("BatchTopic","*");
        //指定每次可以消费10条消息，默认为1
        consumer.setConsumeMessageBatchMaxSize(10);
        //指定每次可以从broker拉取40条消息，默认为32
        consumer.setPullBatchSize(40);
        //注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(msg);
                }
                //消费成功的返回结果
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                //消费异常时的返回结果,RECONSUME_LATER:稍后重新消费
                //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        consumer.start();
        System.out.println("消费者启动了---");
    }
}

六、消息过滤：

学新通
1.tag过滤：

学新通
2.sql过滤：

3.代码举例：

定义tag过滤producer:

com.fan.filter.FilterByTagProducer:

package com.fan.filter;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class FilterByTagProducer {
    public static void main(String[] args)throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("pg");
        producer.setNamesrvAddr("rocketmqOS:9876");
        producer.start();
        //发送的消息均包含tag,为一下三种tag之一
        String[] tags = {"myTagA","myTagB","myTagC"};
        for (int i = 0; i < 10; i  ) {
            byte[] body = ("hi,"   i).getBytes();
            String tag = tags[i % tags.length];
            Message msg = new Message("myTopic0", tag, body);
            SendResult sendResult = producer.send(msg);
            System.out.println(sendResult);
        }
        producer.shutdown();
    }
}

定义tag过滤consumer:
com.fan.filter.FilterByTagConsumer :

package com.fan.filter;
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.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;

public class FilterByTagConsumer {
    public static void main(String[] args)throws  Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
        consumer.setNamesrvAddr("rocketmqOS:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("myTopic","myTagA || myTagB");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt me : msgs) {
                    System.out.println(me);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("消费者已经开始消费---");
    }
}

启动生产者，然后启动消费者：
发现过滤除了只有二级分类标签的myTagA 或者 myTagB：

学新通

定义SQL过滤producer:
学新通

学新通
定义tag过滤consumer:

学新通

运行测试：
学新通

学新通

关闭并修改配置文件：
首先关闭服务器：
Shutdown Servers：

sh bin/mqshutdown broker
sh bin/mqshutdown namesrv

学新通

然后修改配置文件：vim conf/broker.conf

学新通

增加的配置为：

enablePropertyFilter = true

学新通
重新启动服务器：

nohup sh bin/mqnamesrv &
nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &

学新通

然后重新运行SQL过滤的代码看结果：
学新通

七、消息发送重试机制：

学新通

八、消费重试机制：

学新通

九、死信队列：

学新通

这篇好文章是转载于：学新通技术网

RocketMQ应用

一、普通消息：

1.消息发送分类：

（1）同步发送消息：

（2）异步发送消息：

（3）单向发送消息：

2.普通消息代码举例：

定义生产者：

（1）同步发消息生产者：Sync

（2）异步消息发送生产者：Async

（3）单向消息发送生产者：

定义消费者：

二、顺序消息：

三、延迟消息：

四、事务消息：

五、批量消息：

六、消息过滤：

七、消息发送重试机制：

八、消费重试机制：

九、死信队列：

photoshop保存的图片太大微信发不了怎么办

Android 11 保存文件到外部存储，并分享文件

word里面弄一个表格后上面的标题会跑到下面怎么办

《学习通》视频自动暂停处理方法

photoshop扩展功能面板显示灰色怎么办

微信公众号没有声音提示怎么办

excel下划线不显示怎么办

怎样阻止微信小程序自动打开

excel打印预览压线压字怎么办

TikTok加速器哪个好免费的TK加速器推荐