Rocket MQ簡介

時間 2019-11-17

標籤 rocket 簡介简体版

原文原文鏈接

1、概念html

1. 中間件：位於系統之間的服務服務器

2. 消息中間件：消息隊列MQ，用於接收消息、存儲消息、轉發消息的中間件架構

3. Rocket MQ: 分佈式的消息中間件，生產者、消費者、隊列均可以分佈式分佈式

4. 基於Netty開發ide

2、RocketMQ使用spa

1. 在服務器上安裝Rocket MQ.net

2. 啓動rocket mq，即name server，啓動以後監聽端口，等待broker\producer\consumer鏈接code

3. 啓動broker, 設置對應的name server，broker用於收取和存儲消息server

4. 手動/自動建立Topichtm

5. 消費者代碼

public class Consumer {
 
    public static void main(String[] args) throws InterruptedException, MQClientException {
    
        //聲明並初始化一個consumer
        //須要一個consumer group名字做爲構造方法的參數，這裏爲consumer1
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");
 
        //一樣也要設置NameServer地址
        consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
 
        //這裏設置的是一個consumer的消費策略
        //CONSUME_FROM_LAST_OFFSET 默認策略，從該隊列最尾開始消費，即跳過歷史消息
        //CONSUME_FROM_FIRST_OFFSET 從隊列最開始開始消費，即歷史消息（還儲存在broker的）所有消費一遍
        //CONSUME_FROM_TIMESTAMP 從某個時間點開始消費，和setConsumeTimestamp()配合使用，默認是半個小時之前
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 
        //設置consumer所訂閱的Topic和Tag，*表明所有的Tag
        consumer.subscribe("TopicTest", "*");
 
        //設置一個Listener，主要進行消息的邏輯處理
        consumer.registerMessageListener(new MessageListenerConcurrently() {
 
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
 
                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
                
                //返回消費狀態
                //CONSUME_SUCCESS 消費成功
                //RECONSUME_LATER 消費失敗，須要稍後從新消費
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
 
        //調用start()方法啓動consumer
        consumer.start();
 
        System.out.println("Consumer Started.");
    }
}

6. 生產者代碼

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
 
        //聲明並初始化一個producer
        //須要一個producer group名字做爲構造方法的參數，這裏爲producer1
        DefaultMQProducer producer = new DefaultMQProducer("producer1");
        
        //設置NameServer地址,此處應改成實際NameServer地址，多個地址之間用；分隔
        //NameServer的地址必須有，可是也能夠經過環境變量的方式設置，不必定非得寫死在代碼裏
        producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
        
        //調用start()方法啓動一個producer實例
        producer.start();
 
        //發送10條消息到Topic爲TopicTest，tag爲TagA，消息內容爲「Hello RocketMQ」拼接上i的值
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = new Message("TopicTest",// topic
                        "TagA",// tag
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                );
                
                //調用producer的send()方法發送消息
                //這裏調用的是同步的方式，因此會有返回結果
                SendResult sendResult = producer.send(msg);
                
                //打印返回結果，能夠看到消息發送的狀態以及一些相關信息
                System.out.println(sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
 
        //發送完消息以後，調用shutdown()方法關閉producer
        producer.shutdown();
    }
}