【資料圖】

下面是一個完整的示例，演示如何使用 Spring Cloud Bus 在微服務(wù)之間傳遞消息。該示例包含兩個微服務(wù)，一個是消息發(fā)送者，另一個是消息接收者。

首先，我們需要創(chuàng)建一個 Spring Boot 項(xiàng)目，并添加 Spring Cloud Bus 和 RabbitMQ 的依賴。在 pom.xml 文件中添加以下依賴：

    org.springframework.cloud    spring-cloud-starter-bus-amqp    2.2.1.RELEASE    org.springframework.boot    spring-boot-starter-web    2.2.1.RELEASE    org.springframework.boot    spring-boot-starter-amqp    2.2.1.RELEASE

在創(chuàng)建的項(xiàng)目中，我們定義了一個消息類型 MyMessage，該類型包含一個字符串類型的 content 屬性。

javaCopy codepublic class MyMessage implements Serializable {    private static final long serialVersionUID = 1L;    private String content;    public MyMessage(String content) {        this.content = content;    }    public String getContent() {        return content;    }    public void setContent(String content) {        this.content = content;    }}

然后，我們創(chuàng)建了一個消息發(fā)送者，該發(fā)送者使用 Spring Cloud Bus 發(fā)送消息。

@RestControllerpublic class MyController {    @Autowired    private MessageSender messageSender;    @PostMapping("/send")    public void sendMessage(@RequestBody MyMessage message) {        messageSender.sendMessage(message);    }}@Servicepublic class MessageSender {    private static final Logger LOGGER = LoggerFactory.getLogger(MessageSender.class);    @Autowired    private MessageChannel output;    public void sendMessage(MyMessage message) {        LOGGER.info("Sending message: {}", message.getContent());        output.send(MessageBuilder.withPayload(message).build());    }}@Servicepublic class MessageListener {    private static final Logger LOGGER = LoggerFactory.getLogger(MessageListener.class);    @StreamListener(target = Sink.INPUT, condition = "headers["type"]=="MyMessage"")    public void handleMessage(MyMessage message) {        LOGGER.info("Received message: {}", message.getContent());    }}

在這個例子中，我們創(chuàng)建了一個 MyController 類，該類定義了一個發(fā)送消息的 API 接口，接收一個 MyMessage 類型的參數(shù)，將參數(shù)傳遞給 MessageSender 類的 sendMessage() 方法。

MessageSender 類定義了一個 sendMessage() 方法，該方法使用 Spring Cloud Stream 的 output 消息通道發(fā)送消息。

MessageListener 類定義了一個 handleMessage() 方法，該方法使用 @StreamListener 注解監(jiān)聽 Spring Cloud Stream 的 input 消息通道，并根據(jù)消息類型過濾消息。當(dāng)有符合條件的消息到達(dá)時，handleMessage() 方法會被自動調(diào)用，處理接收到的消息。

在以上代碼中，我們使用了 @Autowired 注解自動注入了 MessageSender 和 MessageListener 類，這是 Spring Boot 自帶的依賴注入功能。

運(yùn)行應(yīng)用程序后，我們可以使用 Postman 工具或其他 HTTP 工具發(fā)送 HTTP POST 請求，將消息發(fā)送到消息發(fā)送者的 API 接口，如下所示：

POST http://localhost:8080/send HTTP/1.1Content-Type: application/json{    "content": "Hello, World!"}

當(dāng)消息到達(dá)時，消息接收者會打印消息內(nèi)容，如下所示：

2023-04-19 09:24:47.836  INFO 29740 --- [afka-listener-1] com.example.demo.M

標(biāo)簽：