文章详情

登录

短信预约信息系统项目管理师 报名、考试、查分时间动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

Spring Cloud Alibaba（五）RocketMQ 异步通信实现

2014-06-01 10:31

关注

本文探讨如何使用 RocketMQ Binder 完成 Spring Cloud 应用消息的订阅和发布。

介绍

RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。

RocketMQ 是阿里巴巴在2012年开源的分布式消息中间件，目前已经捐赠给 Apache 软件基金会，并于2017年9月25日成为 Apache 的顶级项目。作为经历过多次阿里巴巴双十一这种“超级工程”的洗礼并有稳定出色表现的国产中间件，以其高性能、低延时和高可靠等特性近年来已经也被越来越多的国内企业使用。

RocketMQ特点

Spring Cloud Stream

Spring Cloud Stream 是一个构建消息驱动微服务的框架。

Spring Cloud Stream 提供了消息中间件配置的统一抽象，推出了 pub/sub，consumer groups，semantics，stateful partition 这些统一的模型支持。

Spring Cloud Stream 核心构件有：Binders、Bindings和Message，应用程序通过 inputs 或者 outputs 来与 binder 交互，通过我们配置来 binding ，而 binder 负责与中间件交互，Message为数据交换的统一数据规范格式。

Binding 在消息中间件与应用程序提供的 Provider 和 Consumer 之间提供了一个桥梁，实现了开发者只需使用应用程序的 Provider 或 Consumer 生产或消费数据即可，屏蔽了开发者与底层消息中间件的接触。

比如 Kafka 的实现 KafkaMessageChannelBinder，RabbitMQ 的实现 RabbitMessageChannelBinder 以及 RocketMQ 的实现 RocketMQMessageChannelBinder。

比如消息 Messaging 对应的模型就包括一个消息体 Payload 和消息头 Header。

spring-cloud-stream 官网

Window搭建部署RocketMQ

下载

当前最新版本为4.6.0

下载出来解压到：D:ocketmq 目录，目录最好不要带空格和太深，否则服务运行可能会报错

启动NameServer服务

在启动之前需要配置系统环境，不然会报错。

Please set the ROCKETMQ_HOME variable in your environment!

系统环境变量名：ROCKETMQ_HOME

根据你解压的目录配置环境变量，比如我的变量值为：D:ocketmq

进入window命令窗口，进入D:ocketmqin目录下，执行

start mqnamesrv.cmd

如上则NameServer启动成功。使用期间，窗口不要关闭。

启动Broker服务

进入bin目录下，输入

start mqbroker.cmd -n localhost:9876

如上的 ip+port 是rocketmq的服务地址和端口。

运行如上命令，可能会报如下错误。找不到或无法加载主类

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

保存再次执行如上命令。执行成功后，提示boot success 代表成功。

示例

本示例实现三种消息的发布以及订阅接收。

创建 RocketMQ 消息生产者

创建 ali-rocketmq-producer 工程，端口为：28081




    
        cloud-alibaba
        com.easy
        1.0.0
    

    4.0.0
    ali-rocketmq-producer
    jar

    

        
        
            com.alibaba.cloud
            spring-cloud-starter-stream-rocketmq
        

        
        
            org.springframework.boot
            spring-boot-starter-web
        

        
            org.springframework.boot
            spring-boot-starter-actuator
        

    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin

application.yml配置

server:
  port: 28081

spring:
  application:
    name: ali-rocketmq-producer
  cloud:
    stream:
      rocketmq:
        binder:
          # RocketMQ 服务器地址
          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);
    }
}

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;

    
    public void send(String msg) {
        Message message = MessageBuilder.withPayload(msg)
                .build();
        source.output1().send(message);
    }

    
    public void sendWithTags(String msg, String tag) {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TAGS, tag)
                .build();
        source.output1().send(message);
    }

    
    public  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 "字符串消息发送成功!";
    }

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

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

创建 RocketMQ 消息消费者

创建 ali-rocketmq-consumer 工程，端口为：28082




    
        cloud-alibaba
        com.easy
        1.0.0
    

    4.0.0

    ali-rocketmq-consumer
    jar

    
        
            com.alibaba.cloud
            spring-cloud-starter-stream-rocketmq
        

        
            org.springframework.boot
            spring-boot-starter-web
        

        
            org.springframework.boot
            spring-boot-starter-actuator
        

    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin

-配置 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 服务地址
        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);
    }
}

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);
    }
}

使用示例

示例关联项目

本示例我们创建了两个项目实现

运行示例测试

首先要启动ali-rocketmq-producer服务及ali-rocketmq-consumer服务

查看服务消费者控制台，输出

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

表示字符串消费成功被input1消费了

查看服务消费者控制台，输出

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消费了，因为input1也订阅了test-topic1，并且没有我们没有加tag过滤，默认表示接收所有消息，所以也能成功接收tagyuntian字符串

查看服务消费者控制台，输出

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却没有输出消息，这是因为sendObject发布的消息走的是test-topic2消息管道，所以不会发布给input1及input2订阅者

资料

阅读原文内容投诉

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）

    难度     813人已做
    查看

  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看

  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看

  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）

    难度     435人已做
    查看

  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机

Spring Cloud Alibaba（五）RocketMQ 异步通信实现

数据库2014-06-01

Vue Axios的异步通信怎么实现

数据库2023-06-29

如何实现AJAX简单异步通信

数据库2024-04-02

C++ BoostAsyncSocket如何实现异步反弹通信

数据库2023-07-05

C++ BoostAsyncSocket实现异步反弹通信的案例详解

数据库2023-03-19

Golang与RabbitMQ实现多服务之间的异步通信

数据库2023-10-08

python多进程及通信实现异步任务的方法

数据库2024-04-02

Java ActiveMQ：轻松实现分布式系统的异步通信

Java ActiveMQ：轻松实现分布式系统的异步通信
数据库2024-02-08

Golang RabbitMQ: 实现异步通信和系统解耦的最佳方案

数据库2023-10-08

如何在 Linux 环境下使用 Spring 框架实现 PHP 同步通信？

如何在 Linux 环境下使用 Spring 框架实现 PHP 同步通信？
数据库2023-10-29

Golang RabbitMQ: 实现异步通信、系统解耦和高性能的最佳方案

数据库2023-10-09

.NET编程——利用C#实现TCP协议的异步通信Socket套接字(WinForm)

数据库2023-09-04
位置：首页-资讯-数据库
咦！没有更多了？去看看其它编程学习网 内容吧
首页课程
资料下载问答资讯