【Spring Cloud】分布式必学springcloud（十二）——结合SpringCloud Stream 异步调用

迈不过友情╰ 2022-05-24 02:42 367阅读 0赞

# 一、前言 #

在上一篇博客中，小编向大家介绍了一下SpringCloud Stream，相信实践过的老铁们，肯定是能收发消息了。但是有一个问题，比如我的发送端，使用的是`@InboundChannelAdapter(value = Source.OUTPUT)`，会一直的向消息队列中发消息。在真正业务场景中，基本不会出现。下面小编就想大家介绍一下，真正的项目中通过发布事件来触发发送消息到消息队列中。

# 二、什么是事件？ #

事件，Event。

当一个事件发生的时候，如果有人关注这个事件，那么他会做出一定的动作。比如，假定七夕情人节马上就要到了，这个就是一个事件，关注着这个节日的情侣们，必定会导致旅馆宾馆的房源紧张，同时也会促进某种产品的销量。

在代码中呢，当我们处理完一个充值后，我们需要向客户发送一个通知，告诉他，充值完毕。充值完成后，就是发生的时间，发送通知是要做的动作，而且这个动作是异步完成的。这种场景是非常常见的。

在springboot中，提供了监听事件 ApplicationListener 和 ApplicationEvent 。

（1）自定义事件：继承ApplicationEvent

（2）定义事件监听器：实现ApplicationListener

（3）使用容器发布事件

# 三、异步发送 #

## 3.1 自定义事件 ##

package com.zero.event;
    
    import lombok.Data;
    import lombok.EqualsAndHashCode;
    import lombok.ToString;
    import org.springframework.context.ApplicationEvent;
    
    /** * 异步通知事件 */
    @Data
    @ToString(callSuper = true)
    @EqualsAndHashCode(callSuper = false)
    public class FeeResponseEvent extends ApplicationEvent { 
    
        public FeeResponseEvent(Object source) {
            super(source);
        }
    }

## 3.2 定义事件监听器：实现ApplicationListener ##

package com.zero.event;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.messaging.Source;
    import org.springframework.context.ApplicationListener;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Component;
    
    /** * Created by Ares on 2018/4/18. */
    @Component
    @EnableBinding(Source.class)
    public class Sender implements ApplicationListener<FeeResponseEvent> { 
    
        @Autowired
        private MessageChannel output;
    
    
        /** * 发送事件消息到消息队列中-王雷-2018年5月18日16:20:22 * @param event 费用事件 */
        private void sendMessageToWorker(FeeResponseEvent event){
            output.send(MessageBuilder.withPayload(event.getSource()).build());
            System.out.println("发送成功："+event.getSource());
    
        }
    
        /** * 当有发布事件发生时，执行此方法-王雷-2018年5月18日16:19:52 * @param feeResponseEvent 费用自定义事件 */
        @Override
        public void onApplicationEvent(FeeResponseEvent feeResponseEvent) {
            this.sendMessageToWorker(feeResponseEvent);
        }
    }

补充：ApplicationListener支持泛型，上面的代码是泛型的写法，下面的是非泛型的写法。

@Component
    //这里注意我们直接把监听类注册成组件
    public class MessageListener implements ApplicationListener { 
    
          private final Logger logger = LoggerFactory.getLogger(MessageListener.class);
    
          @Autowired
          private APIMessageService apiMessageService;
    
    
          @Async
          @Override
          public void onApplicationEvent(ApplicationEvent applicationEvent) {
    
            if (applicationEvent instanceof ApplyEvent) {
                //做处理
            }  
        }
    }

## 3.3 发布事件 ##

@Autowired
        ApplicationContext context;
         /** * 发布事件-王雷-2018年5月18日16:18:56 */
        @Override
        public void feeMessage() {
            FeeResponseEvent event = new FeeResponseEvent("ares");
            context.publishEvent(event);
        }

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

## 3.4 使用spring cloud stream 发送消息 ##

建立消息发送器：

package com.zero.service;
    
    import org.springframework.cloud.stream.annotation.Output;
    import org.springframework.messaging.MessageChannel;
    
    /** * 费用事件发送器 */
    public interface SinkSender { 
        String FEE="fee";
        /** * 费用计算完成事件消息 * @return */
        @Output(FEE)
        MessageChannel feeMessage();
    
    
    }

## 3.5 配置文件配置 ##

spring:
      rabbitmq:
        host: 1*2.3*1.**.2
        port: 5672
        username: guest
        password: guest
      cloud:
        stream:
          bindings:
            output:
              destination: fee
              contentType: 'application/json'

## 3.6 发送测试 ##

启动服务后，调用发布事件，会引起事件监听监听到，然后会向mq中放数据。

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

## 3.7 使用 spring cloud stream 接收消息 ##

配置监听器：

配置监听器，一直监听Exchange名字为fee的。当监听到有消息的时候，就会调用下面指定的方法进行相关的操作。

package com.zero.service;
    
    import org.springframework.cloud.stream.annotation.Input;
    import org.springframework.messaging.MessageChannel;
    
    /** * Created by Ares on 2018/5/18. */
    public interface SinkIn { 
        String INPUT = "fee";
    
        @Input("fee")
        MessageChannel input();
    }

package com.zero.listener;
    
    import com.zero.service.SinkIn;
    import org.springframework.cloud.stream.annotation.EnableBinding;
    import org.springframework.cloud.stream.annotation.StreamListener;
    import org.springframework.stereotype.Component;
    
    /** * 接收队列消息 * @author Ares * */
    @Component
    @EnableBinding(SinkIn.class)
    public class ReceiverMessage { 
    
        @StreamListener(SinkIn.INPUT)
        public void receive(Object playload) {
    
            System.out.println("接收消息:" + playload);
    
        }
    
    }

## 3.8 异步设置 ##

启异步配置在配置类中添加注解@EnableAsync

package com.zero;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
    import org.springframework.scheduling.annotation.EnableAsync;
    
    @SpringBootApplication
    @EnableDiscoveryClient
    @EnableAsync
    public class FeeMasterWebApplication {
    
        private static final Logger log = LoggerFactory.getLogger(FeeMasterWebApplication.class);
    
        public static void main(String[] args) {
            log.info("费用模块提供者启动....");
            SpringApplication.run(FeeMasterWebApplication.class, args);
            log.info("费用模块提供者启动成功....");
    
        }
    }

通过MessageListener 在onApplicationEvent上添加注解@Async，@Async也可以添加在类上。

/** * 当有发布事件发生时，执行此方法-王雷-2018年5月18日16:19:52 * @param feeResponseEvent 费用自定义事件 */
        @Override
        @Async
        public void onApplicationEvent(FeeResponseEvent feeResponseEvent) {
            this.sendMessageToWorker(feeResponseEvent);
        }
    }

# 四、小结 #

通过这次的使用，有一种思想可以提一下，在做架构设计的时候，必定会涉及到异步操作，所以我们可以把设计到异步操作的提取出来，单独做一个服务。这样就可以解耦合。