Apache RocketMQ：使用官方demo测试rocketmq

Posted 2021-11-05 你是小KS

当前rocketmq版本4.9

1. 声明

当前内容主要为使用官方的demo测试之前的rocketmq是否正常，测试发送和消费消息，主要参考官方文档

2. pom依赖

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

启动rocketmq的两个进程使用&符号

3.demo

同步消息生产者

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new
            DefaultMQProducer("testGroup");
        // Specify name server addresses.
        producer.setNamesrvAddr("192.168.1.101: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 class Consumer {

    public static void main(String[] args) throws InterruptedException, MQClientException {

        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("testGroup");
         
        // Specify name server addresses.
        consumer.setNamesrvAddr("192.168.1.101: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);
                for (MessageExt messageExt : msgs) {
                	try {
						String msg = new String(messageExt.getBody(),RemotingHelper.DEFAULT_CHARSET).intern();
						System.out.println("Receive New Messages: "+msg);
					} catch (UnsupportedEncodingException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //Launch the consumer instance.
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}

结果：

4. 出现的问题

会在初次启动的过程中疯狂的创建index里面的文件—>2000多个，但是关闭后再次启动却不增加

5.总结

1.rocketmq还是和其他的模型差不多，都有消费者，topic，订阅，ack确认