SpringCloud学习笔记(五)-SpringCloudStream集成kafka(下)

简介: Spring Cloud Stream是构建消息驱动的微服务应用程序框架。提供统一的接收发送管道以连接到消息代理。通过@EnableBinding注解开启SpringCloudStream的支持。通过@StreamListener注解,使其接收流处理的时间。

五、发送消息到输出通道


/**
 * kafka消息发送器
 * @author dbq
 * @date 2019/9/26 17:50
 */
@Component
public class EsKafkaMessageSender {
    @Autowired
    private EsChannel channel;
    /**
     * 消息发送到默认通道:缺省通道对应缺省主题
     * @param message
     */
    public void sendToDefaultChannel(String message){
        channel.sendEsDefaultMessage().send(MessageBuilder.withPayload(message).build());
    }
    /**
     * 消息发送到告警通道:告警通道对应告警主题
     * @param message
     */
    public void sendToAlarmChannel(String message){
        channel.sendEsAlarmMessage().send(MessageBuilder.withPayload(message).build());
    }
}


注入先前定义的通道EsChannel,sendToDefaultChannel、sendToAlarmChannel分别为我们自定义的两个发送方法,可将消息发送到不同的通道中,每个通道对应一个kafka的主题。


六、从输入通道订阅消息


@EnableBinding(value = EsChannel.class)
public class EsStreamListener {
    /**
     * 从缺省通道接收消息
     * @param message
     */
    @StreamListener(EsChannel.ES_DEFAULT_INPUT)
    public void receive(Message<String> message){
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        System.out.println(sdf.format(new Date())+"------start--------安全用电默认消息:" + message);
        try {
            Thread.sleep(1000*10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(sdf.format(new Date())+"------end--------安全用电默认消息");
    }
    /**
     * 从告警通道接收消息
     * @param message
     */
    @StreamListener(EsChannel.ES_ALARM_INPUT)
    public void receiveAlarm(Message<String> message){
        System.out.println("订阅告警消息:" + message);
    }
}


从不同的通道实现消息的订阅。


七、这样完整的消息系统就搭建好了,定义Controller发送消息测试


@ApiOperation(value = "test1", httpMethod = "POST")
    @PostMapping(value = "/test1", produces = "application/json;charset=UTF-8")
    public void test1(String message, HttpServletRequest request,
                             HttpServletResponse response) {
        sender.sendToDefaultChannel(message);
        sender.sendToDefaultChannel(message);
        sender.sendToDefaultChannel(message);
        sender.sendToDefaultChannel(message);
    }
    @ApiOperation(value = "test", httpMethod = "POST")
    @PostMapping(value = "/test2", produces = "application/json;charset=UTF-8")
    public void test2(String message, HttpServletRequest request,
                      HttpServletResponse response) {
        sender.sendToAlarmChannel(message);
    }


test1:发送消息的缺省消息通道


test2:发送消息到告警消息通道


八、并发性测试


如七中所示,一次发送4条消息到缺省消息通道中,并启动两个实例(即两个微服务组成一个小型集群),在并发性配置为1的情况下,即spring.cloud.stream.bindings.es_default_input.consumer.concurrency=1


实例1

2019-09-30 11:13:14------start--------默认消息...
2019-09-30 11:13:24------end--------默认消息


实例2

2019-09-30 11:13:14------start--------默认消息:...
2019-09-30 11:13:24------end--------默认消息
2019-09-30 11:13:24------start--------默认消息:...
2019-09-30 11:13:34------end--------默认消息
2019-09-30 11:13:34------start--------默认消息:...
2019-09-30 11:13:44------end--------默认消息


通过打印日志(日志做了简化处理)可以看出,两个实例之间是做到了并发消费,但是在1个实例内部,并没有并发消费。


如果将concurrency修改为2.


日志如下


实例1

2019-09-30 11:31:13------start--------:...
2019-09-30 11:31:13------start--------默认消息:...
2019-09-30 11:31:23------end--------默认消息
2019-09-30 11:31:23------end--------默认消息
2019-09-30 11:31:23------start--------默认消息:...
2019-09-30 11:31:33------end--------默认消息


实例2

2019-09-30 11:31:13------start--------默认消息:...
2019-09-30 11:31:23------end--------


从日志可以看出,实例1中实现了两个线程的并发消费。


相关文章
|
4天前
|
消息中间件 Java Kafka
Springboot集成高低版本kafka
Springboot集成高低版本kafka
|
4天前
|
SpringCloudAlibaba Java 网络架构
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(二)Rest微服务工程搭建
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(二)Rest微服务工程搭建
66 0
|
4天前
|
负载均衡 Java 网络架构
在SpringCloud2023中快速集成SpringCloudGateway网关
本文主要简单介绍SpringCloud2023实战中SpringCoudGateway的搭建。后续的文章将会介绍在微服务中使用熔断Sentinel、鉴权OAuth2、SSO等技术。
41 2
在SpringCloud2023中快速集成SpringCloudGateway网关
|
4天前
|
SpringCloudAlibaba Java 网络架构
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(七)Spring Cloud Gateway服务网关
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(七)Spring Cloud Gateway服务网关
125 0
|
4天前
|
SQL 数据库 Windows
SpringCloud集成seata分布式事务控制
SpringCloud集成seata分布式事务控制
20 0
|
4天前
|
SpringCloudAlibaba 负载均衡 Java
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(目录大纲)
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(目录大纲)
74 1
|
4天前
|
Java Nacos Sentinel
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(九)Nacos+Sentinel+Seata
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(九)Nacos+Sentinel+Seata
240 0
|
4天前
|
消息中间件 SpringCloudAlibaba Java
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(八)Config服务配置+bus消息总线+stream消息驱动+Sleuth链路追踪
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(八)Config服务配置+bus消息总线+stream消息驱动+Sleuth链路追踪
795 0
|
4天前
|
SpringCloudAlibaba Java 测试技术
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(六)Hystrix(豪猪哥)的使用
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(六)Hystrix(豪猪哥)的使用
50 1
|
4天前
|
SpringCloudAlibaba 负载均衡 Java
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(五)OpenFeign的使用
【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记(五)OpenFeign的使用
51 0

热门文章

最新文章