rocketmq精读-白红宇

rocketmq精读

阅读量：4144 次

发布时间：2019-05-25

本文共 6099 字，大约阅读时间需要 20 分钟。

rocketmq简单使用

1.项目介绍

2.结构组件及工作流程

3.基本使用

4.各种功能

5.可靠性优先

6.吞吐量优先

7.其它

1.项目介绍

rocketmq是由阿里开源的Apache中间件顶级项目.经历过多次阿里双十一生产验证.

项目地址:

2.结构组件

Producer:消息生产者

Consumer:消息消费者

Broker:消息暂存

NameServer:路由信息存储

消息组件:Topic Queue Tag Key

Topic: 消息分类

Queue: 逻辑上消息存储在队列里

Tag: 消息标签

Key: 消息标识

工作流程:

首先咱们启动NameServer,Broker.然后可以在broker上创建Topic并指定Queue数量.

生产者连接到NameServer上获取路由信息,发送消息到broker

消费者连接到NmaeServer上获取路由信息,注册监听器,拉取消息,消费

3.基本使用

3.1同步发送

public class SyncProducer {    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();    }}

3.2异步发送

public class AsyncProducer {    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();    }}

3.3one-way发送

public class OnewayProducer {    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();    }}

3.4消息消费

public class Consumer {    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
   
     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");    }}

总结一下:

生产者步骤:

创建生产者->设置NameServer->启动生产者->发送消息->关闭生产者

消费者步骤:

创建消费者->设置NameServer->订阅topic->注册监听服务->启动消费者