3. springboot集成rocketmq

青旅半醒 2023-07-01 10:53 27阅读 0赞

# 引入依赖 #

这里我们使用`rocketmq-spring-boot-starter`来访问rocketmq。

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.0.4</version>
    </dependency>

配置文件中需要配置rocketmq的nameserver等信息：

server:
      port: 8090
    
    rocketmq:
      name-server: node1:9876
      producer:
        group: myGroupName

# 测试发送消费消息 #

@RestController
    @Slf4j
    public class HelloController { 
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        @GetMapping("/hello")
        public String hello() { 
            // 生产者
            rocketMQTemplate.convertAndSend("TopicTest", "Hello, World!");
            rocketMQTemplate.send("TopicTest", MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
    
            return "Welcome to springboot2 world ~";
        }
    }
    
    
    //消费者
    @Slf4j
    @Service
    @RocketMQMessageListener(topic = "TopicTest", consumerGroup = "my-consumer_TopicTest")
    public class MyConsumer implements RocketMQListener<String> { 
        public void onMessage(String message) { 
            log.info("received message: " + message);
        }
    }
    
    
    //启动类
    @SpringBootApplication
    public class Application { 
    
        public static void main(String[] args) { 
            SpringApplication.run(Application.class, args);
        }
    }

启动服务，调用`http://127.0.0.1:8090/hello`来发送消息。  
可以看到在控制台输出：

2020-01-21 11:54:39.392  INFO 1436 --- [MessageThread_1] com.example.service.MyConsumer           : received message: Hello, World!
    2020-01-21 11:54:39.410  INFO 1436 --- [MessageThread_2] com.example.service.MyConsumer           : received message: Hello, World! I'm from spring message

然后在rocketmq的控制台也同样定义消息，可以看到数据更加详细信息：

[root@node1 bin]# ./tools.sh org.apache.rocketmq.example.quickstart.Consumer
    11:27:15.018 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
    Consumer Started.
    
    
    ConsumeMessageThread_5 Receive New Messages: [MessageExt [queueId=3, storeSize=290, queueOffset=250, sysFlag=0, bornTimestamp=1579578879352, bornHost=/192.168.229.1:53977, storeTimestamp=1579578879475, storeHost=/192.168.229.101:10911, msgId=C0A8E56500002A9F0000000000038054, commitLogOffset=229460, bodyCRC=1816839120, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=251, CONSUME_START_TIME=1579578879517, id=d5fa0a8c-b86a-4495-ccc2-da178a2c2ce3, UNIQ_KEY=AC10728F059C2437C6DC698D78DF0000, CLUSTER=DefaultCluster, WAIT=false, contentType=text/plain;charset=UTF-8, timestamp=1579578879032}, body=[72, 101, 108, 108, 111, 44, 32, 87, 111, 114, 108, 100, 33], transactionId='null'}]] 
    ConsumeMessageThread_6 Receive New Messages: [MessageExt [queueId=0, storeSize=314, queueOffset=250, sysFlag=0, bornTimestamp=1579578879360, bornHost=/192.168.229.1:53977, storeTimestamp=1579578879481, storeHost=/192.168.229.101:10911, msgId=C0A8E56500002A9F0000000000038176, commitLogOffset=229750, bodyCRC=1978678191, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=251, CONSUME_START_TIME=1579578879525, id=5f89df97-d11a-6508-99fc-1f85491cffaf, UNIQ_KEY=AC10728F059C2437C6DC698D79800002, CLUSTER=DefaultCluster, WAIT=false, contentType=text/plain;charset=UTF-8, timestamp=1579578879359}, body=[72, 101, 108, 108, 111, 44, 32, 87, 111, 114, 108, 100, 33, 32, 73, 39, 109, 32, 102, 114, 111, 109, 32, 115, 112, 114, 105, 110, 103, 32, 109, 101, 115, 115, 97, 103, 101], transactionId='null'}]]