【消息队列】消息中间件RocketMQ4.X急速入门2

//设置生产者发送broker失败重复发送的次数
producer.setRetryTimesWhenSendFailed(5);

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

//设置广播模式，默认集群模式，广播模式下失败不会重新消费，继续消费下一个
consumer.setMessageModel(MessageModel.BROADCASTING);

//默认是同步发送
SendResult sendResult = payProducer.getProducer().send(message);

//
SendResult sendResult = payProducer.getProducer().send(message,new SendCallback(){
    //发送成功处理函数
    onSuccess(){}
    //异常处理函数
    onException(){}
});

//单向发送
payProducer.getProducer().sendOneway(message);

producer.getProducer().send(message,new MessageQueueSelector(){
    select(List<MessageQueue> mqs, Message msg, Object arg){
        Integer queueNum = (Integer)arg;
        return mqs.get(queueNum);
      }
},0);

RestController
@RequestMapping("api/v1/pay")
public class PayController {
    @Autowired
    private PayProducer payProducer;
    @RequestMapping("/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(JmsConfig.TOPIC,"taga","6688",text.getBytes());
        /**
         * 异步发送到指定队列
         */
        payProducer.getProducer().send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                int queueNum = Integer.parseInt(arg.toString());
                return mqs.get(queueNum);
            }
        }, 3, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult);
            }
            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        });
        /**
         * 同步发送到指定队列
         */
       /* SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                //查找队列
                int queueNum = Integer.parseInt(arg.toString());
                return mqs.get(queueNum);
            }
        },0);
        System.out.println(sendResult);*/
        /**
         * 单向发送
         */
     /*   message.setDelayTimeLevel(3);
        SendResult sendResult = payProducer.getProducer().send(message);
        //payProducer.getProducer().sendOneway(message);
        System.out.println(sendResult);*/
        return new HashMap<>();
    }
}

public MessageQueue select(List<MessageQueue> mqs,Message msg,Object arg){
    //订单id进行取模
    Long orderId  = Integer.parseLong(arg.toString());
    long index = orderId%mqs.size();
    return mqs.get((int)index);
}

@RestController
@RequestMapping("api/v1/pay")
public class PayController {
    @Autowired
    private PayProducer payProducer;
    @RequestMapping("/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        List<ProducerOrder> producerList = ProducerOrder.getProducerList();
        for (ProducerOrder producerOrder : producerList) {
            Message message = new Message(JmsConfig.TOPIC,"",producerOrder.getOrderId()+"",producerOrder.toString().getBytes());
            SendResult send = payProducer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long orderId = (Long)arg;
                    //订单号和队列集合长度取模
                    long index = orderId % mqs.size();
                    return mqs.get((int)index);
                }
            },producerOrder.getOrderId());
            MessageQueue messageQueue = send.getMessageQueue();
            int queueId = messageQueue.getQueueId();
            System.out.println("orderId:"+producerOrder.getOrderId()+"---orderType:"+producerOrder.getType()+"---queueId:"+queueId);
            System.out.println("---------------------------");
        }
        return new HashMap<>();
    }
}

concurrently多线程消费
@Component
public class PayConsumer {
    private DefaultMQPushConsumer consumer;
    private String consumerGroup = "pay_consumer_group";
    public PayConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.subscribe("","*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    MessageExt message = msgs.get(0);
                    int times = message.getReconsumeTimes();
                    try{
//                        if(message.getKeys().equalsIgnoreCase("6688")){
//                            throw new Exception();
//                        }
                        String body = new String(msgs.get(0).getBody());
                        String tags = msgs.get(0).getTags();
                        String keys = msgs.get(0).getKeys();
                        System.out.println("message:"+message+"--body:"+body+"--tags:"+tags+"--keys:"+keys);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }catch (Exception e){
                        if(times>=3){
                            System.out.println("消息失败，转入人工处理");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        System.out.println("消息异常，重新投递");
                        System.out.println("消息重投次数："+times);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });
            consumer.start();
            System.out.println("consumer.start()...");
    }
}

orderly单线程消费（保证顺序消费必须用单线程消费）
@Component
public class PayOrderlyConsumer {
    private DefaultMQPushConsumer consumer;
    private String consumerGroup = "pay_orderly_consumer_group";
    public PayOrderlyConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.subscribe(JmsConfig.TOPIC,"*");
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext consumeConcurrentlyContext) {
                    MessageExt message = msgs.get(0);
                    int times = message.getReconsumeTimes();
                    try{
                        String body = new String(msgs.get(0).getBody());
                        String tags = msgs.get(0).getTags();
                        String keys = msgs.get(0).getKeys();
                        System.out.println("message:"+message+"--body:"+body+"--tags:"+tags+"--keys:"+keys);
                        return ConsumeOrderlyStatus.SUCCESS;
                    }catch (Exception e){
                        System.out.println("消息异常，重新投递");
                        System.out.println("消息重投次数："+times);
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                }
            });
            consumer.start();
            System.out.println("consumer.start()...");
    }
}

The broker does not support consumer to filter message by SQL92
broker.conf配置文件中配置
enablePropertyFilter=true
备注，修改之后要重启Broker