RocketMq-Spring 请求应答语义支持

红太狼 2021-06-24 13:57 784阅读 0赞

RocketMQ-Spring 提供 请求/应答 语义支持。

*  Producer端

发送Request消息使用SendAndReceive方法

> 注意
> 
> 同步发送需要在方法的参数中指明返回值类型
> 
> 异步发送需要在回调的接口中指明返回值类型

@SpringBootApplication
    public class ProducerApplication implements CommandLineRunner{
        @Resource
        private RocketMQTemplate rocketMQTemplate;
        
        public static void main(String[] args){
            SpringApplication.run(ProducerApplication.class, args);
        }
        
        public void run(String... args) throws Exception {
            // 同步发送request并且等待String类型的返回值
            String replyString = rocketMQTemplate.sendAndReceive("stringRequestTopic", "request string", String.class);
            System.out.printf("send %s and receive %s %n", "request string", replyString);
    
            // 异步发送request并且等待User类型的返回值
            rocketMQTemplate.sendAndReceive("objectRequestTopic", new User("requestUserName",(byte) 9), new RocketMQLocalRequestCallback<User>() {
                @Override public void onSuccess(User message) {
                    System.out.printf("send user object and receive %s %n", message.toString());
                }
    
                @Override public void onException(Throwable e) {
                    e.printStackTrace();
                }
            }, 5000);
        }
        
        @Data
        @AllArgsConstructor
        public class User implements Serializable{
            private String userName;
        		private Byte userAge;
        }
    }

*  Consumer端

需要实现RocketMQReplyListener<T, R> 接口，其中T表示接收值的类型，R表示返回值的类型。

@SpringBootApplication
    public class ConsumerApplication{
        
        public static void main(String[] args){
            SpringApplication.run(ConsumerApplication.class, args);
        }
        
        @Service
        @RocketMQMessageListener(topic = "stringRequestTopic", consumerGroup = "stringRequestConsumer")
        public class StringConsumerWithReplyString implements RocketMQReplyListener<String, String> {
            @Override
            public String onMessage(String message) {
              System.out.printf("------- StringConsumerWithReplyString received: %s \n", message);
              return "reply string";
            }
          }
       
        @Service
        @RocketMQMessageListener(topic = "objectRequestTopic", consumerGroup = "objectRequestConsumer")
        public class ObjectConsumerWithReplyUser implements RocketMQReplyListener<User, User>{
            public void onMessage(User user) {
              	System.out.printf("------- ObjectConsumerWithReplyUser received: %s \n", user);
              	User replyUser = new User("replyUserName",(byte) 10);	
              	return replyUser;
            }
        }
    
        @Data
        @AllArgsConstructor
        public class User implements Serializable{
            private String userName;
        		private Byte userAge;
        }
    }