Rocket MQ的基本应用

Rocket MQ的基本应用

    消息队列,常用于应用间通信。

    本篇文章基于RocketMQ官方文档

相关概念和组件

    概念

        组件

    Name Server如果出现宕机,当前Producer和Consumer已经获取的topic路由信息会保持,既有功能不受影响,但若是新增topic和Broker,将会无法传达消息

特点

  1. 先进先出

    队列的特点,先进先出,不受人为的干预

   2.发布-订阅

   3.持久化

   4.分布式

    MQ的定义就是多应用并发通信的高性能中间件。

应用场景

    应用解耦

    应用间通信不使用MQ会发生并发问题,同时可能会消耗应用资源,影响性能。

    

    通知

    应用b一直在监听MQ中来自应用a的消息,逐条消费。

    应用分发

    MQ是一对多的,可以在多应用间分发(广播)消息。

    限制流量

    具有队列的特性,这就说明MQ不存在并发问题,同时有且仅有一条数据被消费。

    分布式事务

RocketMQ

    项目中使用了RocketMQ,这个mq中间件只提供了按topic进行发布-订阅的通信方式,并未提供点对点的消息队列。

    官网文档入口

    涉及的一些概念:

(1)NameServer:在MQ集群中做的是做命名服务,更新和路由发现 broker服务; (2)Broker-Master:broker 消息主机服务器; (3)Broker-Slave:broker 消息从机服务器; (4)Producer:消息生产者; (5)Consumer:消息消费者;

开始使用RocketMQ         

        quick-start:发送与接收消息

        RocketMQ有三种消息发送方式,示例代码均来自于官网。

public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
            DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " +
                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
}
public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        for (int i = 0; i < 100; i++) {
                final int index = i;
                //Create a message instance, specifying topic, tag and message body.
                Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                producer.send(msg, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.printf("%-10d OK %s %n", index,
                            sendResult.getMsgId());
                    }
                    @Override
                    public void onException(Throwable e) {
                        System.out.printf("%-10d Exception %s %n", index, e);
                        e.printStackTrace();
                    }
                });
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
public static void main(String[] args) throws Exception{
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // Specify name server addresses.
        producer.setNamesrvAddr("localhost:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest" /* Topic */,
                "TagA" /* Tag */,
                ("Hello RocketMQ " +
                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            producer.sendOneway(msg);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }

        接收消息:

public static void main(String[] args) throws InterruptedException, MQClientException {
        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
         
        // Specify name server addresses.
        consumer.setNamesrvAddr("localhost:9876");
        
        // Subscribe one more more topics to consume.
        consumer.subscribe("TopicTest", "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //Launch the consumer instance.
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }

        消息的顺序性保证:Selector

        首先需要了解的是,对于部署了多个队列(多个broker)的topic消息,在选择消息队列时采用的是随机算法(或hash、或随机、或依赖时间戳)在消费消息时会轮询Queue列表消费,这样能最大限度的做到负载均衡,但并不能保证消息的顺序性,或者说在同一个topic多节点部署broker时,当消息量很大时,大概率消息的消费顺序是与消息生产不一致的。但有时很多时候消息要保证传输的顺序性,即FIFO,例如支付流程中先后发出的支付请求和支付成功请求。producer可以通过调用

 send(Message msg, MessageQueueSelector selector, Object arg)

在发送消息时指定selector,从而按指定的规则选择发送消息的通道(队列)。

        MessageQueueSelector是函数式接口:

public interface MessageQueueSelector {
    MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}

        其中mqs就是消息队列的列表,msg是消息体,arg是自定义的参数。

        通过继承该接口定义新的队列选择器,发送消息:

 public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        MQProducer producer = new DefaultMQProducer("example_group_name");
        //Launch the instance.
        producer.start();
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 100; i++) {
            int orderId = i % 10;
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTestjjj", tags[i % tags.length], "KEY" + i,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                Integer id = (Integer) arg;
                int index = id % mqs.size();
                return mqs.get(index);
            }
            }, orderId);

            System.out.printf("%s%n", sendResult);
        }
        //server shutdown
        producer.shutdown();
    }

        同样,消费端也需要做相应的处理

//TODO

参考链接:MQ(消息队列)常见的应用场景解析


2019-06-28鱼鱼

{{blog.title}}

创建于 {{blog.createTimeStr}}   created  by  {{blog.author}} {{tag}}
最后修改于 {{blog.timelineStr}}
修改文档