精通springcloud：自定义RabbitMQ代理的连接

Posted 2021-06-26 jinggege795

自定义与RabbitMQ代理的连接

在前面的章节中，已经介绍了使用Docker镜像启动RabbitMQ代理的方法，因而有必要记住这个命令。它将启动一个带RabitMQ的独立Docker容器，可在端口5672下使用，其用户界面Web控制台可在端口15672 下使用。

docker run -d --name rabbit -p 15672:15672 -P 5672:5672 rabbi tnq : management

应使用application.yml文件中的spring.rabbit. *属性覆盖默认的RabbitMQ地址。

spring:

rabbi tmq:

host: 192.168.99.100

port: 5672

默认情况下，Spring Cloud Stream会为通信创建主题交换信息。这种类型的交换更适合发布/订阅交互模型。开发人员也可以使用exchangeType 属性覆盖它，就像在application.yml 的片段中一样，如下所示。

spring:

cloud:

stream:

rabbit:

bindings:

output:

producer:

exchangeTyper direct

input:

consumer:

exchangeType: direct

开发人员应该为order-service服务和account-service服务提供相同的配置设置。这里不必手动创建任何交换信息。如果它不存在，则它会在启动期间由应用程序自动创建。否则，应用程序只会绑定到该交换信息。默认情况下，它创建交换信息时所使用的名称，对于@Input通道就是输入的名称，对于@Output通道就是输出的名称。这些名称可以用springcloud.stream.bindings.output.destination和

spring.cloud.stream.bindings.input.destination属性覆盖，其中的输入和输出都是通道的名称。此配置选项不仅是Spring Cloud Stream功能的“个很好的补充，而且是用于关联服务间通信中的输入和输出目标的键值设置。

 

要解释为什么会发生这种情况也非常简单。在我们的示例中，一方面， order-service 是消息源应用程序，因而它会将消息发送到输出通道。然后，另一方面，account-service 服务将侦听输入通道上的传入消息。如果order- service服务的输出通道和account-service服务的输入通道未引用代理上的相同目标，则它们之间的通信将失败。总之，我们决定使用名称为order-out和orders-in的目的地，并且为order-service服务提供了以下配置。

spring:

cloud :

stream:

bindings :

output:

destination: orders -out

input :

destination: orders-in

account- service服务的类似配置设置则刚好相反。

spring:

cloud:

stream:

bindings:

output:

destination: orders-in

input :

destination: orders-out

在两个应用程序都启动之后，可以使用其Web管理控制台轻松查看RabbitMQ代理上声明的交换列表，该控制台位于htp://192.68.99.100:15672 (guestguest) 。如图11.2所示的是隐式创建的交换信息(Exchange)，开发人员可能会看到出于测试目的创建的两个目标。

 

默认情况下，Spring Cloud Stream将提供一个输入和一个输出消息通道。可以想象一种情况，即我们的系统需要为每种类型的消息通道提供多个目的地。现在可以回到示例系统架构，并考虑每个订单由另外两个微服务异步处理的情况。到目前为止，只有account-service 服务一直在监听来自order-service 服务的传入事件。在当前示例中，product-service服务将是传入订单的接收者，它在这种情况下的主要目标是管理可用产品的数量，并根据订单细节减少它们。它要求我们在order-service服务中定义两个输入和输出消息通道，因为我们仍然有基于直接RabbitMQ交换信息的点对点通信，其中每个消息都可以由一个使用者处理。

 

在这种情况下，我们应该使用@Input和@Output方法声明两个接口。每个方法都必须返回一个channel对象。Spring Cloud Stream提供两个可绑定的消息组件一用于出站通信的MessageChannel,以及它的扩展——用于入站通信的SubscribableChannel.这是与product-service服务交互的接口定义。已创建用于通过account-service服务进行消息传递的类似接口。

public interface ProductOrder {

@Input

SubscribableChannel productordersIn() ;

@output

MessageChannel productOrdersOut() ;

}

下一步是通过使用@EnableBinding(value- {Accountrder.as,Productorder.cass})来注解其主类以激活应用程序的声明组件。现在，可以使用它们的名称在配置属性中引用这些通道，如

sringcloudstream.bindings.productOrdersOut.destination- product-orders-in。使用@Input和@Output注解时，可以通过指定通道名称来自定义每个通道名称，示例如下。

public interface Productorder {

@Input ("productordersIn")

SubscribableChannel ordersIn() :

@output ("productordersOut")

MessageChannel ordersOut() ;

}

基于自定义接口声明，Spring Cloud Stream将生成实现该接口的bean.但是，仍然必须在负责发送消息的bean中访问它。与前面的示例相比，直接注入绑定通道会更简便。以下是当前产品订单消息发送者的bean实现。还有一个类似的bean实现，它可以将消息发送到account-service.

@Service

public class ProductorderSender {

@Autowired

private MessageChannel output;

@Autowired

public. SendingBean (@Qualifier ("productordersout") MessageChannel

output) {

this.output = output ;

}

public boolean send(order order) {

return this . output . send (MessageBui lder。withPayload (order) . build（）);

}

}

还应为目标服务提供每个消息通道的自定义接口。侦听消息者应绑定到消息代理上的正确消息通道和目标。

@streamListener (ProductOrder . INPUT)

public void receiveOrder (Order order) throws JsonProcessingException {

service.process (order) ;

}

与其他Spring Cloud项目集成

你可能已经注意到，示例系统混合了不同类型的服务间通信。有些微服务使用典型的RESTful HTTP API,还有一些使用消息代理。在单个应用程序中混合不同类型的通信也没有异议。例如，可以使用Spring Cloud Stream将sping-loud-starter-feign包含在项目中，并使用@EnableFeignClients 注解启用它。在我们的示例系统中，这两种不同的通信方式结合了account-service 服务,它通过消息代理与order-service服务集成，并通过RESTAPI与product-service 服务集成。以下是account-service 服务模块中Feign 客户端的product-service服务实现。

@FeignClient (name一"product-service")

public interface ProductClient 4

@PostMapping("/ids")

List<Product> findByIds (@RequestBody List<Long> ids) ;

}

还有其他好消息。由于Spring Cloud Sleuth的存在，在通过网关传入系统的单个请求期间交换的所有消息都具有相同的traceld。无论是同步REST通信还是异步消息传递，开发人员都可以使用标准日志文件或日志聚合器工具(如Elastic Stack)轻松跟踪和关联微服务之间的日志。

 

现在是运行和测试示例系统的好时机。首先，开发人员必须使用mvn clean install命令构建整个项目。要使用两个微服务来侦听两个不同交换上的消息来访问代码示例，开发人员应该切换到advanced分支Spst:/ihbopomin/samomple 

springclou-messagingtree/advanced)。应该启动其中可用的所有应用程序一网关、发现和3个微服务(account-service服务、order-service 服务、product-service 服务)。目前讨论的案例假设我们还使用其Docker容器启动了RabbitMQ、 Logstash、 Elasticsearch 和Kibana.有关如何使用Docker镜像在本地运行ElasticStack的详细说明，请参阅之前文章提到的“分布式日志记录和跟踪”。图11.3 详细显示了本示例系统的架构。

 

运行所有必需的应用程序和工具后，即可继续进行测试。以下是示例请求，它可以通过API网关发送到order-service服务。

curl -H"Content-Type: application/json" -X POST -d

'I"customerId- :1, "produetIds":[1,3.4], "gtatus":"NEN" '

http://localhost: 8080/ap1/order

如果按照前文所述配置的应用程序运行测试时，第一次可能无法正常工作。有些开发人员可能会有点困惑，因为通常它是在默认设置下测试的。要让它正确运行，还必须在application.yml 中添加以下属性: spring. cloud. sreramrabit.binding 

.outputproducer.routingKeyExpession:"#"。它可以将默认生产者的路由键设置为与应用程序引导期间自动创建的交换信息的路由键一致。 在如图11.4所示的屏幕截图中，可能会看到一个输出交换信息定义。

 

在执行上述修改之后，应该能成功完成测试。微服务打印的日志通过traceld相互关联。我们在logback-spring.xml中稍微修改了默认的Sleuth日志记录格式，这就是现在配置的方式——%d {HmssSSS}%-5level [%X {X-B3-Traceld: - } ,%X {X- B3-Spanld:- }] %msg%n。在发送测试性的order-service服务测试请求后，日志将记录以下信息。

12:34:48.696 INFO [ 68038dd653f7b0b, 68038cdd653f7b0b] Order saved:

{"id":1, "status": "NEW", "price" :0, "customerId" :1, "accountId" :null,

"productId s":[1,3,4] }

12:34:49.821 INFO [ 68038cdd653f7b0b, 68038cdd653f7b0b] Order sent:

{"isSent":true }

可以看到，account-service 服务也使用相同的日志记录格式，并打印与order-service服务相同的traceld.

12:34:50.079 INEO [ 68038cdd653f7b0b, 23432d962ec92f7a] Order processed:

("id":1, "status" :"NEW", "price":0, "customerId":1, "accountId":null,

"productId s":[1,3,4]}

12:34:50.332 INFO [ 68038dd653f7b0b, 23432d962ec92f7a] Account found :

("id":1, "number":"1234567890"，"balance" : 50000, "customerId":1}

12:34:52.344 INFO [ 68038cdd653f7b0b, 23432d962ec92f7a] Products found:

[{"id":1, "name":"TestI", "price":1000}, {"id":3, "name":"Test3",

"price":2000}，("id":4, "name" :"Test4" "price":3000} ]

可以使用Elastic Stack聚合在单个事务期间生成的所有日志。可以通过X-B3-TraceId字段(如9dale5c83094390d)过滤条目，如图11.5所示。

 

可以点赞+转发后，下方扫码来获取这套完整的体系资料。