RocketMQ实践之消息收发（二）

拼搏现实的明天。 2022-08-28 06:57 253阅读 0赞

### 一、工程搭建 ###

**前提：**  
请大家自行装好IntelliJ IDEA并配好Maven。

**File->New->Project…**

--------------------

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_19_color_FFFFFF_t_70_g_se_x_16]

**Maven->org.apache.maven.archetypes:maven-archetype-quickstart**

--------------------

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16]  
**name:rocket-mq,GroupId:my.demo**

--------------------

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
**Finish**

--------------------

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
**Add Dependency**

--------------------

maven:

<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.3.0</version>
        </dependency>

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
**如果还是红的，Maven->Reload**

--------------------

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 4]

### 二、消息发送 ###

**同步发送：**

--------------------

package my.demo;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    
    /** * Hello world! * */
    public class SyncProducer 
    { 
        public static void main( String[] args ) throws Exception { 
    
            DefaultMQProducer producer = new
                    DefaultMQProducer("p_group1");
            producer.setNamesrvAddr("localhost:9876");
            producer.start();
            for (int i = 0; i < 100; i++) { 
                Message msg = new Message("TopicTest" /* Topic */,
                        "TagA" /* Tag */,
                        ("Hello RocketMQ " +
                                i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            }
            producer.shutdown();
    
        }
    }

**插曲：**

--------------------

运行报错

D:\dev\Java\jdk1.8.0_261\bin\java.exe "-javaagent:D:\dev\IntelliJ IDEA 2021.2\lib\idea_rt.jar=56818:D:\dev\IntelliJ IDEA 2021.2\bin" -Dfile.encoding=UTF-8 -classpath D:\dev\Java\jdk1.8.0_261\jre\lib\charsets.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\deploy.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\access-bridge-64.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\cldrdata.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\dnsns.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\jaccess.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\jfxrt.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\localedata.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\nashorn.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunec.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunjce_provider.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunmscapi.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\sunpkcs11.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\ext\zipfs.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\javaws.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jce.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jfr.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jfxswt.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\jsse.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\management-agent.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\plugin.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\resources.jar;D:\dev\Java\jdk1.8.0_261\jre\lib\rt.jar;D:\dev\code\koala\rocket-mq\target\classes;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-client\4.3.0\rocketmq-client-4.3.0.jar;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-common\4.3.0\rocketmq-common-4.3.0.jar;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-remoting\4.3.0\rocketmq-remoting-4.3.0.jar;C:\Users\Administrator\.m2\repository\com\alibaba\fastjson\1.2.29\fastjson-1.2.29.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-all\4.0.42.Final\netty-all-4.0.42.Final.jar;C:\Users\Administrator\.m2\repository\org\apache\rocketmq\rocketmq-logging\4.3.0\rocketmq-logging-4.3.0.jar;C:\Users\Administrator\.m2\repository\io\netty\netty-tcnative-boringssl-static\1.1.33.Fork26\netty-tcnative-boringssl-static-1.1.33.Fork26.jar;C:\Users\Administrator\.m2\repository\org\apache\commons\commons-lang3\3.4\commons-lang3-3.4.jar my.demo.App
    Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, TopicTest
    See http://rocketmq.apache.org/docs/faq/ for further details.
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.sendDefaultImpl(DefaultMQProducerImpl.java:610)
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1223)
    	at org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl.send(DefaultMQProducerImpl.java:1173)
    	at org.apache.rocketmq.client.producer.DefaultMQProducer.send(DefaultMQProducer.java:214)
    	at my.demo.App.main(App.java:29)

更换版本成与broker一样的版本4.9.1

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 5]  
控制台打印出发送的消息  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 6]  
**消费消息：**

--------------------

package my.demo;
    
    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.message.MessageExt;
    
    import java.util.List;
    
    public class Consumer { 
    
        public static void main(String[] args) throws InterruptedException, MQClientException { 
    
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("c_group1");
             
            consumer.setNamesrvAddr("localhost:9876");
            
            consumer.subscribe("TopicTest", "*");
            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;
                }
            });
    
            consumer.start();
            System.out.printf("Consumer Started.%n");
        }
    }

**异步发送：**

--------------------

package my.demo;
    
    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;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    public class AsyncProducer { 
        public static void main(String[] args) throws Exception { 
            DefaultMQProducer producer = new DefaultMQProducer("p_group2");
            producer.setNamesrvAddr("localhost:9876");
            producer.start();
            producer.setRetryTimesWhenSendAsyncFailed(0);
            
            int messageCount = 100;
            final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
            for (int i = 0; i < messageCount; i++) { 
                try { 
                    final int index = i;
                    Message msg = new Message("TopicTest",
                        "TagA",
                        "OrderID188",
                        "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
                    producer.send(msg, new SendCallback() { 
                        @Override
                        public void onSuccess(SendResult sendResult) { 
                            countDownLatch.countDown();
                            System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                        }
    
                        @Override
                        public void onException(Throwable e) { 
                            countDownLatch.countDown();
                            System.out.printf("%-10d Exception %s %n", index, e);
                            e.printStackTrace();
                        }
                    });
                } catch (Exception e) { 
                    e.printStackTrace();
                }
            }
            countDownLatch.await(5, TimeUnit.SECONDS);
            producer.shutdown();
        }
    }

**单向消息：**

--------------------

package my.demo;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    public class OnewayProducer { 
        public static void main(String[] args) throws Exception{ 
            DefaultMQProducer producer = new DefaultMQProducer("p_group3");
            producer.setNamesrvAddr("localhost:9876");
            producer.start();
            for (int i = 0; i < 100; i++) { 
                Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                        i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );
                producer.sendOneway(msg);
            }
            Thread.sleep(5000);        
            producer.shutdown();
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_19_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/8d097211a37b45fcb14b37d37f21d4d2.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/5bc2014cd32544bcacefa561e39823db.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/e8280ad54ab74d95a945b456212b0f76.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220828/23d12be1147a4b33b60b9a2a767ea135.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220828/6603c3ecc0694c97950afaea4277b890.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220828/f0333708275146bea69f66077f29c7bd.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 5]: /images/20220828/ef754107f3234e51b9707c1ef6193a18.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA54Gs5a2Q6Z2e6bG85Li2_size_20_color_FFFFFF_t_70_g_se_x_16 6]: /images/20220828/4576067dec36460191cc6b8c9a47b40a.png