iis添加网站 别名,合肥网络公司哪个最好,wordpress板块大小,企业管理系统数据库设计背景及痛点 现如今消息中间件(MQ)在互联网项目中被广泛的应用#xff0c;特别是大数据行业应用的特别的多#xff0c;现在市面上也流行这多个消息中间件框架#xff0c;比如ActiveMQ、RabbitMQ、RocketMQ、Kafka等#xff0c;这些消息中间件各有各的优劣#xff0c;但是想… 背景及痛点 现如今消息中间件(MQ)在互联网项目中被广泛的应用特别是大数据行业应用的特别的多现在市面上也流行这多个消息中间件框架比如ActiveMQ、RabbitMQ、RocketMQ、Kafka等这些消息中间件各有各的优劣但是想要解决的问题都基本相同。由于每个框架都有它自己的使用方式这无疑是增加了开发者的学习成本以及添加相同的业务复杂度。框架的变更或者多个中间件的混合使用使得业务逻辑代码中中间件的切换、项目的维护和开发都会变得更加繁琐。 有没有一种技术让我们不再需要关注具体MQ的使用细节我们只需要专注业务逻辑的开发让程序根据实际项目的使用自己去适配绑定自动在各种MQ内切换呢springcloud stream便为此而生。 关于stream 我们用一句话来描述stream就是:屏蔽底层消息中间件的差异降低切换版本统一消息的编程模型 官方定义SpringCloud Stream是一个构建消息驱动微服务的框架应用通过inputs或者outputs来与SpringCloud Stream中的binder对象交互我们通过配置来绑定消息中间件而SpringCloud Stream的binder对象负责与消息中间件交互所以我们只需要搞清楚如何与SpringCloud Stream交互即可方便的使用消息中间件。 SpringCloud Stream通过Spring Integration来连接消息代理中间件以实现消息事件驱动它提供了个性化的自动化配置引用了发布订阅、消费组、分区的三个核心概念但是目前仅支持RabbitMQ和Kafka 设计思想 在此之前 以前的架构 生产者和消费者通过消息媒介(queue等)传递信息内容(Message)消息必须通过特定的通道MessageChannel通过消息的发布与订阅来决定消息的发送和消费publish/subscrib。 引入中间件 现在假如我们用到了RabbitMQ和Kafka由于这两个消息中间件的架构上的不同像RabbitMQ有Exchange而Kafka有topiche和Partitions分区 引入中间件之后 binder中input对于消费者output对应生产者。 这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰我们如果用了两个消息队列的其中一种,但是后面因为业务需求需要改用另外一种消息队列进行迁移,这时候无疑就是一 个灾难性的一大堆东西都要重新推倒重新做,因为它跟我们的系统耦合了,这时候springcloud Stream给我们提供了一种解耦合的方式。 屏蔽底层差异 在没有绑定器(Builder)这个概念的情况下我们的SpringBoot应用要直接与消息中间件进行信息交互的时候,由于各消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性,通过定义绑定器作为中间件完美地实现了应用程序与消息中间件细节之间的隔离。通过向应用程序暴露统一的Channel通道 使得应用程序不需要再考虑各种不同的消息中间件实现。 通过定义绑定器Binder作为中间层实现了应用程序与消息中间件细节之间的隔离。 处理架构 Stream对消息中间件的进一步封装可以做到代码层面对中间件的无感知甚至于动态的切换中间件(rabbitmq切换为kafka),使得微服务开发的高度解耦服务可以关注更多自己的业务流程。 处理架构 其遵循了发布-订阅模式主要使用的就是Topic主题进行广播RabbitMQ就是Exchange在Kafka中就是Topic 通过定义绑定器Binder作为中间层实现了应用程序与消息中间件细节之间的隔离。 stream流程 stream流程 Binder很方便的连接中间件屏蔽差异 Channel通道是队列Queue的一种抽象在消息通讯系统中就是实现存储和转发的媒介通过对Channel对队列进行配置 Source和Sink简单的可理解为参照对象是Spring Cloud Stream自身从Stream发布消息就是输出接受消息就是输入 常用api和注解 常用api和注解 使用示例 基本环境 注册中心Eureka可以是其他。 消息中间件RabbitMQ rabbitmq:host: localhostport: 5672username: guestpassword: guest生产端 依赖 !--stream rabbit --
dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-stream-rabbit/artifactId
/dependency
!--eureka client--
dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-netflix-eureka-client/artifactId
/dependency
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId
/dependency
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-actuator/artifactId
/dependency配置文件 server:port: 8801spring:application:name: cloud-stream-providercloud:# stream 配置stream:binders: # 配置绑定的消息中间件的服务信息defaultRabbit: # 自定义的一个名称用来下面 bindings 绑定type: rabbit # 消息组件的类型environment: #相关环境配置设置rabbitmq的环境spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服务的整合处理output: # 通道名称destination: testExchange # 定义要使用的Exchange的名称content-type: application/json # 设置消息类型对象为json文本是text/plainbinder: defaultRabbit # 设置要绑定的服务的具体设置就是我们上面配置的defaultRabbiteureka:client:#表示是否将自己注册进EurekaServer默认为trueregister-with-eureka: true#是否从EurekaServer抓取已有的注册消息默认为true单节点无所谓集群必须设置为true才能配合ribbon使用负载均衡fetch-registry: trueservice-url:#单机版defaultZone: http://localhost:8080/eureka/instance:prefer-ip-address: trueinstance-id: sender01定义接口 这里需要定义一个接口并实现它方便其他业务调用。 public interface IMessageProvider {/*** 发送接口* param msg* return*/public String send(String msg);
}接口实现 接口实现中需要添加EnableBinding注解并引入Source.class,为什么引入Source.class呢因为它是生产者我们参考stream流程图就可以知道 import com.martain.study.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import javax.annotation.Resource;EnableBinding(Source.class)
public class MessageProvider implements IMessageProvider {/*** 注入消息发送管道*/Resourceprivate MessageChannel output;Overridepublic String send(String msg) {output.send(MessageBuilder.withPayload(msg).build());System.out.println(******send message:msg);return msg;}
}定义测试controller
RestController
public class TestController {AutowiredIMessageProvider messageProvider;GetMapping(/sendMsg)public String sendMsg(){String msg UUID.randomUUID().toString();return messageProvider.send(msg);}}启动类 SpringBootApplication
public class StreamProviderApplication8801 {public static void main(String[] args) {SpringApplication.run(StreamProviderApplication8801.class,args);}
} 服务启动之后多次请求/sendMsg发送了多条消息。 生产服务生产消息 消费端 依赖 !--stream rabbit --dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-stream-rabbit/artifactId/dependency!--eureka client--dependencygroupIdorg.springframework.cloud/groupIdartifactIdspring-cloud-starter-netflix-eureka-client/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-actuator/artifactId/dependency配置文件 与生产者类似只是bindings中的output改成了input server:port: 8802
spring:application:name: cloud-stream-consumecloud:# stream 配置stream:binders: # 配置绑定的消息中间件的服务信息defaultRabbit: # 自定义的一个名称用来下面 bindings 绑定type: rabbit # 消息组件的类型environment: #相关环境配置设置rabbitmq的环境spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服务的整合处理input: # 通道名称destination: testExchange # 定义要使用的Exchange的名称content-type: application/json # 设置消息类型对象为json文本是text/plainbinder: defaultRabbit # 设置要绑定的服务的具体设置就是我们上面配置的defaultRabbiteureka:client:#表示是否将自己注册进EurekaServer默认为trueregister-with-eureka: true#是否从EurekaServer抓取已有的注册消息默认为true单节点无所谓集群必须设置为true才能配合ribbon使用负载均衡fetch-registry: trueservice-url:#单机版defaultZone: http://localhost:8080/eureka/instance:prefer-ip-address: trueinstance-id: recover01 接收服务 接收服务只需要再类名前添加EnableBinding()注解并引入Sink.class类而实际接收的方法中需要添加StreamListener(Sink.INPUT)注解。 package com.martain.study.controller;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component; Component
EnableBinding(Sink.class)
public class ReceiveMessageListenerController {/*** 获取本服务的端口*/Value(${server.port})private String serverPort;/*** 这里表示监听sink的input* param message*/StreamListener(Sink.INPUT)public void input(MessageString message){System.out.println(**** recv msg :message.getPayload() in port serverPort);}
} 启动类 SpringBootApplication
public class StreamConsumerApplication8802 {public static void main(String[] args) {SpringApplication.run(StreamConsumerApplication8802.class,args);}
} 启动生产服务后在启动消费服务多次请求生产服务发送消息我们可以发现消费者能很快的消费这些消息。 消费者消费消息 消息分组 当我们有多个消费者时,这个时候生产者生产一条消息会发现所有的消费者都会消费这个消息。比如在一些订单系统的场景中如果一个订单被多个处理服务一起获取到就容易造成数据错误那我们如何避免这种情况呢这时我们就可以使用Stream的消息分组来解决重复消费问题。 如何实现Stream的消息分组呢我们只要简单的在yml文件中配置spring.cloud.stream.bindings.input.group即可。示例如下 ...
spring:application:name: cloud-stream-consumecloud:# stream 配置stream:binders: # 配置绑定的消息中间件的服务信息defaultRabbit: # 自定义的一个名称用来下面 bindings 绑定type: rabbit # 消息组件的类型environment: #相关环境配置设置rabbitmq的环境spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服务的整合处理input: # 通道名称destination: testExchange # 定义要使用的Exchange的名称content-type: application/json # 设置消息类型对象为json文本是text/plainbinder: defaultRabbit # 设置要绑定的服务的具体设置就是我们上面配置的defaultRabbitgroup: groupA # 配置分组...如果没有设置该属性当消费服务启动时会有个随机的组名。 如果我们将所有的消费服务的group熟悉都设置成一致的话这些服务就会在同一个组里面从而能够保证消息只被应用消费一次。 同一组的消费者是竞争关系不可以重复消费。 消息持久化 当生产者在持续生产消息消费服务突然挂了使得拥有许多消息并没有被消费如果消费没有配置分组的话消费服务重启是无法消费未消费的消息的如果配置了分组的话当消费服务重启之后可以自动去消费未消费的数据。 © 著作权归作者所有,转载或内容合作请联系作者 喜欢的朋友记得点赞、收藏、关注哦
