久久久伊人热,亚洲成人综合社区

本文探討如何使用 RocketMQ Binder 完成 Spring Cloud 應(yīng)用消息的訂閱和發(fā)布。

介紹

RocketMQ 是一款開源的分布式消息系統(tǒng)，基于高可用分布式集群技術(shù)，提供低延時(shí)的、高可靠的消息發(fā)布與訂閱服務(wù)，廣泛應(yīng)用于多個(gè)領(lǐng)域，包括異步通信解耦、企業(yè)解決方案、金融支付、電信、電子商務(wù)、快遞物流、廣告營銷、社交、即時(shí)通信、移動應(yīng)用、手游、視頻、物聯(lián)網(wǎng)、車聯(lián)網(wǎng)等。

RocketMQ 是阿里巴巴在2012年開源的分布式消息中間件，目前已經(jīng)捐贈給 Apache 軟件基金會，并于2017年9月25日成為 Apache 的頂級項(xiàng)目。作為經(jīng)歷過多次阿里巴巴雙十一這種“超級工程”的洗禮并有穩(wěn)定出色表現(xiàn)的國產(chǎn)中間件，以其高性能、低延時(shí)和高可靠等特性近年來已經(jīng)也被越來越多的國內(nèi)企業(yè)使用。

RocketMQ特點(diǎn)

是一個(gè)隊(duì)列模型的消息中間件，具有高性能、高可靠、高實(shí)時(shí)、分布式等特點(diǎn)
Producer、Consumer、隊(duì)列都可以分布式
Producer 向一些隊(duì)列輪流發(fā)送消息，隊(duì)列集合稱為 Topic，Consumer 如果做廣播消費(fèi)，則一個(gè) Consumer 實(shí)例消費(fèi)這個(gè) Topic 對應(yīng)的所有隊(duì)列，如果做集群消費(fèi)，則多個(gè) Consumer 實(shí)例平均消費(fèi)這個(gè) Topic 對應(yīng)的隊(duì)列集合
能夠保證嚴(yán)格的消息順序
支持拉（pull）和推（push）兩種消息模式
高效的訂閱者水平擴(kuò)展能力
實(shí)時(shí)的消息訂閱機(jī)制
億級消息堆積能力
支持多種消息協(xié)議，如 JMS、OpenMessaging 等
較少的依賴

Spring Cloud Stream

Spring Cloud Stream 是一個(gè)構(gòu)建消息驅(qū)動微服務(wù)的框架。

Spring Cloud Stream 提供了消息中間件配置的統(tǒng)一抽象，推出了 pub/sub，consumer groups，semantics，stateful partition 這些統(tǒng)一的模型支持。

Spring Cloud Stream 核心構(gòu)件有：Binders、Bindings和Message，應(yīng)用程序通過 inputs 或者 outputs 來與 binder 交互，通過我們配置來 binding ，而 binder 負(fù)責(zé)與中間件交互，Message為數(shù)據(jù)交換的統(tǒng)一數(shù)據(jù)規(guī)范格式。

Binding: 包括 Input Binding 和 Output Binding。

Binding 在消息中間件與應(yīng)用程序提供的 Provider 和 Consumer 之間提供了一個(gè)橋梁，實(shí)現(xiàn)了開發(fā)者只需使用應(yīng)用程序的 Provider 或 Consumer 生產(chǎn)或消費(fèi)數(shù)據(jù)即可，屏蔽了開發(fā)者與底層消息中間件的接觸。

Binder: 跟外部消息中間件集成的組件，用來創(chuàng)建 Binding，各消息中間件都有自己的 Binder 實(shí)現(xiàn)。

比如 Kafka 的實(shí)現(xiàn) KafkaMessageChannelBinder，RabbitMQ 的實(shí)現(xiàn) RabbitMessageChannelBinder 以及 RocketMQ 的實(shí)現(xiàn) RocketMQMessageChannelBinder。

Message：是 Spring Framework 中的一個(gè)模塊，其作用就是統(tǒng)一消息的編程模型。

比如消息 Messaging 對應(yīng)的模型就包括一個(gè)消息體 Payload 和消息頭 Header。

spring-cloud-stream 官網(wǎng)

Window搭建部署RocketMQ

下載

當(dāng)前最新版本為4.6.0

下載出來解壓到：D:\rocketmq 目錄，目錄最好不要帶空格和太深，否則服務(wù)運(yùn)行可能會報(bào)錯(cuò)

啟動NameServer服務(wù)

在啟動之前需要配置系統(tǒng)環(huán)境，不然會報(bào)錯(cuò)。

Please set the ROCKETMQ_HOME variable in your environment!

系統(tǒng)環(huán)境變量名：ROCKETMQ_HOME

根據(jù)你解壓的目錄配置環(huán)境變量，比如我的變量值為：D:\rocketmq

進(jìn)入window命令窗口，進(jìn)入D:\rocketmq\bin目錄下，執(zhí)行

start mqnamesrv.cmd

如上則NameServer啟動成功。使用期間，窗口不要關(guān)閉。

啟動Broker服務(wù)

進(jìn)入bin目錄下，輸入

start mqbroker.cmd -n localhost:9876

如上的 ip+port 是rocketmq的服務(wù)地址和端口。

運(yùn)行如上命令，可能會報(bào)如下錯(cuò)誤。找不到或無法加載主類

如果出此情況，打開bin-->runbroker.cmd，修改%CLASSPATH%成"%CLASSPATH%"

保存再次執(zhí)行如上命令。執(zhí)行成功后，提示boot success 代表成功。

示例

本示例實(shí)現(xiàn)三種消息的發(fā)布以及訂閱接收。

創(chuàng)建 RocketMQ 消息生產(chǎn)者

創(chuàng)建 ali-rocketmq-producer 工程，端口為：28081

pom.xml添加依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">

    <parent>
        <artifactId>cloud-alibaba</artifactId>
        <groupId>com.easy</groupId>
        <version>1.0.0</version>
    </parent>

    <modelVersion>4.0.0</modelVersion>
    <artifactId>ali-rocketmq-producer</artifactId>
    <packaging>jar</packaging>

    <dependencies>

        <!--rocketmq依賴-->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>

        <!--web依賴-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

配置 Output 的 Binding 信息并配合 @EnableBinding 注解使其生效

application.yml配置

server:
  port: 28081

spring:
  application:
    name: ali-rocketmq-producer
  cloud:
    stream:
      rocketmq:
        binder:
          # RocketMQ 服務(wù)器地址
          name-server: 127.0.0.1:9876
      bindings:
        output1: {destination: test-topic1, content-type: application/json}
        output2: {destination: test-topic2, content-type: application/json}

management:
  endpoints:
    web:
      exposure:
        include: '*'
  endpoint:
    health:
      show-details: always

ArProduceApplication.java

@SpringBootApplication
@EnableBinding({MySource.class})
public class ArProduceApplication {

    public static void main(String[] args) {
        SpringApplication.run(ArProduceApplication.class, args);
    }
}

消息生產(chǎn)者服務(wù)

MySource.java

package com.easy.arProduce;

import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;

public interface MySource {

    @Output("output1")
    MessageChannel output1();

    @Output("output2")
    MessageChannel output2();
}

SenderService.java

package com.easy.arProduce;

import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.MimeTypeUtils;

@Service
public class SenderService {

    @Autowired
    private MySource source;

    /**
     * 發(fā)送字符串
     *
     * @param msg
     */
    public void send(String msg) {
        Message message = MessageBuilder.withPayload(msg)
                .build();
        source.output1().send(message);
    }

    /**
     * 發(fā)送帶tag的字符串
     *
     * @param msg
     * @param tag
     */
    public void sendWithTags(String msg, String tag) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TAGS, tag)
                .build();
        source.output1().send(message);
    }

    /**
     * 發(fā)送對象
     *
     * @param msg
     * @param tag
     * @param <T>
     */
    public <T> void sendObject(T msg, String tag) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TAGS, tag)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build();
        source.output2().send(message);
    }
}

編寫 TestController.java 控制器方便測試

package com.easy.arProduce;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(value = "test")
public class TestController {
    @Autowired
    SenderService senderService;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public String send(String msg) {
        senderService.send(msg);
        return "字符串消息發(fā)送成功!";
    }

    @RequestMapping(value = "/sendWithTags", method = RequestMethod.GET)
    public String sendWithTags(String msg) {
        senderService.sendWithTags(msg, "tagStr");
        return "帶tag字符串消息發(fā)送成功!";
    }

    @RequestMapping(value = "/sendObject", method = RequestMethod.GET)
    public String sendObject(int index) {
        senderService.sendObject(new Foo(index, "foo"), "tagObj");
        return "Object對象消息發(fā)送成功!";
    }
}

創(chuàng)建 RocketMQ 消息消費(fèi)者

創(chuàng)建 ali-rocketmq-consumer 工程，端口為：28082

pom.xml添加依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">

    <parent>
        <artifactId>cloud-alibaba</artifactId>
        <groupId>com.easy</groupId>
        <version>1.0.0</version>
    </parent>

    <modelVersion>4.0.0</modelVersion>

    <artifactId>ali-rocketmq-consumer</artifactId>
    <packaging>jar</packaging>

    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

-配置 Input 的 Binding 信息并配合 @EnableBinding 注解使其生效

application.yml配置

server:
  port: 28082

spring:
  application:
    name: ali-rocketmq-consumer
  cloud:
    stream:
      rocketmq:
        binder:
          name-server: 127.0.0.1:9876 #rocketmq 服務(wù)地址
        bindings:
          input1: {consumer.orderly: true}  #是否排序
          input2: {consumer.tags: tagStr}   #訂閱 帶tag值為tagStr的字符串
          input3: {consumer.tags: tagObj}   #訂閱 帶tag值為tabObj的字符串
      bindings:
        input1: {destination: test-topic1, content-type: text/plain, group: test-group1, consumer.maxAttempts: 1}
        input2: {destination: test-topic1, content-type: application/plain, group: test-group2, consumer.maxAttempts: 1}
        input3: {destination: test-topic2, content-type: application/plain, group: test-group3, consumer.maxAttempts: 1}

management:
  endpoints:
    web:
      exposure:
        include: '*'
  endpoint:
    health:
      show-details: always

ArConsumerApplication.java

package com.easy.arConsumer;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;

@SpringBootApplication
@EnableBinding({MySource.class})
public class ArConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(ArConsumerApplication.class, args);
    }
}

消息消費(fèi)者服務(wù)

MySource.java

package com.easy.arConsumer;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface MySource {
    @Input("input1")
    SubscribableChannel input1();

    @Input("input2")
    SubscribableChannel input2();

    @Input("input3")
    SubscribableChannel input3();
}

ReceiveService.java

package com.easy.arConsumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

@Service
@Slf4j
public class ReceiveService {

    @StreamListener("input1")
    public void receiveInput1(String receiveMsg) {
        log.info("input1 接收到了消息：" + receiveMsg);
    }

    @StreamListener("input2")
    public void receiveInput2(String receiveMsg) {
        log.info("input2 接收到了消息：" + receiveMsg);
    }

    @StreamListener("input3")
    public void receiveInput3(@Payload Foo foo) {
        log.info("input3 接收到了消息：" + foo);
    }
}

使用示例

示例關(guān)聯(lián)項(xiàng)目

本示例我們創(chuàng)建了兩個(gè)項(xiàng)目實(shí)現(xiàn)

ali-rocketmq-producer：RocketMQ 消息服務(wù)生產(chǎn)者，服務(wù)名：ali-rocketmq-producer，端口：28081
ali-rocketmq-consumer：RocketMQ 消息服務(wù)消費(fèi)者，服務(wù)名：ali-rocketmq-producer，端口：28082

運(yùn)行示例測試

首先要啟動ali-rocketmq-producer服務(wù)及ali-rocketmq-consumer服務(wù)

訪問消息服務(wù)生產(chǎn)者地址： http://localhost:28081/test/send?msg=yuntian

查看服務(wù)消費(fèi)者控制臺，輸出

2019-12-04 15:37:47.859  INFO 6356 --- [MessageThread_1] com.easy.arConsumer.ReceiveService       : input1 接收到了消息：yuntian
2019-12-04 15:37:47.859  INFO 6356 --- [MessageThread_1] s.b.r.c.RocketMQListenerBindingContainer : consume C0A8096E200818B4AAC212CDA70E0014 cost: 1 ms

表示字符串消費(fèi)成功被input1消費(fèi)了

訪問消息服務(wù)生產(chǎn)者地址： http://localhost:28081/test/sendWithTags?msg=tagyuntian

查看服務(wù)消費(fèi)者控制臺，輸出

2019-12-04 15:38:09.586  INFO 6356 --- [MessageThread_1] com.easy.arConsumer.ReceiveService       : input2 接收到了消息：tagyuntian
2019-12-04 15:38:09.592  INFO 6356 --- [MessageThread_1] com.easy.arConsumer.ReceiveService       : input1 接收到了消息：tagyuntian
2019-12-04 15:38:09.592  INFO 6356 --- [MessageThread_1] s.b.r.c.RocketMQListenerBindingContainer : consume C0A8096E200818B4AAC212CDFCD30015 cost: 6 ms

表示帶tag的字符串成功被input2和input1消費(fèi)了，因?yàn)閕nput1也訂閱了test-topic1，并且沒有我們沒有加tag過濾，默認(rèn)表示接收所有消息，所以也能成功接收tagyuntian字符串

訪問消息服務(wù)生產(chǎn)者地址： http://localhost:28081/test/sendObject?index=1

查看服務(wù)消費(fèi)者控制臺，輸出

2019-12-04 15:41:15.285  INFO 6356 --- [MessageThread_1] com.easy.arConsumer.ReceiveService       : input3 接收到了消息：Foo{id=1, bar='foo'}

表示input3成功接收到了tag帶tagObj的對象消息了，而input1卻沒有輸出消息，這是因?yàn)閟endObject發(fā)布的消息走的是test-topic2消息管道，所以不會發(fā)布給input1及input2訂閱者

Spring Boot、Cloud 學(xué)習(xí)項(xiàng)目

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

Spring Cloud Alibaba（五）RocketMQ 異步通信實(shí)現(xiàn)

Spring Cloud Alibaba（五）RocketMQ 異步通信實(shí)現(xiàn)

介紹

RocketMQ特點(diǎn)

Spring Cloud Stream

Window搭建部署RocketMQ

下載

啟動NameServer服務(wù)

啟動Broker服務(wù)

示例

創(chuàng)建 RocketMQ 消息生產(chǎn)者

創(chuàng)建 RocketMQ 消息消費(fèi)者

使用示例

示例關(guān)聯(lián)項(xiàng)目

運(yùn)行示例測試

資料

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九 欧美,1769亚洲,黄色成人av

Spring Cloud Alibaba（五）RocketMQ 異步通信實(shí)現(xiàn)

介紹

RocketMQ特點(diǎn)

Spring Cloud Stream

Window搭建部署RocketMQ

下載

啟動NameServer服務(wù)

啟動Broker服務(wù)

示例

創(chuàng)建 RocketMQ 消息生產(chǎn)者

創(chuàng)建 RocketMQ 消息消費(fèi)者

使用示例

示例關(guān)聯(lián)項(xiàng)目

運(yùn)行示例測試

資料

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av