专栏首页品茗ITSpringBoot入门建站全系列(十八)整合RabbitMQ(AMQP类消息队列)

SpringBoot入门建站全系列(十八)整合RabbitMQ(AMQP类消息队列)

SpringBoot入门建站全系列(十八)整合RabbitMQ(AMQP类消息队列)

一、概述

1.1 简介

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间的通信。对于消息中间件,常见的角色大致也就有Producer(生产者)、Consumer(消费者)

常见的消息中间件产品:

(1)ActiveMQ

ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。这里介绍的是ActiveMQ的使用。

(2)RabbitMQ

AMQP协议的领导实现,支持多种场景。淘宝的MySQL集群内部有使用它进行通讯,OpenStack开源云平台的通信组件,最先在金融行业得到运用。

(3)ZeroMQ

史上最快的消息队列系统

(4)Kafka

Apache下的一个子项目 。特点:高吞吐,在一台普通的服务器上既可以达到10W/s的吞吐速率;完全的分布式系统。适合处理海量数据。

Jms

JMS即Java消息服务(Java Message Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。Java消息服务是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持。

AMQP

AMQP是一种协议,更准确的说是一种binary wire-level protocol(链接协议)。这是其和JMS的本质差别,AMQP不从API层进行限定,而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。意味着我们可以使用Java的AMQP provider,同时使用一个python的producer加一个rubby的consumer。从这一点看,AQMP可以用http来进行类比,不关心实现的语言,只要大家都按照相应的数据格式去发送报文请求,不同语言的client均可以和不同语言的server链接。

代码可以在SpringBoot组件化构建https://www.pomit.cn/java/spring/springboot.html中的RabbitMQ组件中查看,并下载。

本篇将通过介绍RabbitMQ的三种模式的配合使用来介绍下Springboot如何整合RabbitMQ。

**如果大家正在寻找一个java的学习环境,或者在开发中遇到困难,可以<a

href="https://jq.qq.com/?_wv=1027&k=52sgH1J"

target="_blank">

加入我们的java学习圈,点击即可加入

</a>

,共同学习,节约学习时间,减少很多在学习中遇到的难题。**

1.2 RabbitMQ基础概念

说基础概念,这里就不多废话了,想必大家都已经百度过了,我这里只说下关于本篇的用到的一些概念。

几个概念:queue、exchange、Binding、routingKey:

queue:队列,和用什么模式无关。

exchange:交换机,就是个代理,决定你用哪种模式,每个模式需要定义一个exchange。

Binding:队列绑定,需要将queue与exchange绑定,可以交叉绑定,一个Binding,就代表一个路径,比如queue绑定到direct模式的exchange,这个queue就可以接收direct模式的消息;queue绑定到topic模式的exchange,这个queue就可以接收topic模式的消息。

routingKey:路由key,等于一条消息的归类。通过这个匹配到与exchange绑定的queue。因为绑定的时候是指定了routingKey的类型。

三种模式:direct、fanout、topic。不是说只有这三种,是这里只说这三种常见的。

direct模式: 一对一模式,发送路径是固定的,一个routingKey只会转发到绑定到指定exchange并绑定到exchange的queue。

fanout模式: 一对多,没有routingKey,消息发送的时候发送到exchange,因为没有routingKey,exchange会转发到绑定到该exchange的所有queue。

topic模式:一对多,可以绑定多个key,也可以绑定模式匹配的key。消息发送的时候发送到exchange,exchange会根据绑定的所有key和模式转发到相应的queue。所以这个模式叫做订阅。

二、配置

本文假设你已经引入spring-boot-starter-web。已经是个SpringBoot项目了,如果不会搭建,可以打开这篇文章看一看《SpringBoot入门建站全系列(一)项目建立》

2.1 Maven依赖

使用RabbitMQ可以使用spring-boot-starter-amqp,方便快捷,一般springboot对大多数开源项目都做了整合,提供了专用的stater。

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

2.2 配置文件

在application.properties 中需要配置RabbitMQ的信息,也可以配置自定义的配置,如:

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

rabbit.direct.queue=spring.queue.direct
rabbit.direct.exchange=spring.exchange.direct
rabbit.direct.key=spring.key.direct

rabbit.common.queue=spring.queue.common
rabbit.common.queue.pattern=spring.key.*

rabbit.common.queue2=spring.queue.common2
rabbit.common.queue2.pattern=*.key.topic

rabbit.fanout.exchange=spring.exchange.fanout

rabbit.topic.exchange=spring.exchange.topic

rabbit.topic.key=spring.key.topic

这里面,

  • spring.rabbitmq.*是springboot自动装配的配置,rabbitmq的相关信息。
  • rabbit.direct.*是我自定义的配置,是direct模式下需要的配置。
  • rabbit.fanout.*是我自定义的配置,是fanout模式下需要的配置。
  • rabbit.topic.*是我自定义的配置,是topic模式下需要的配置。
  • rabbit.common.*是我自定义的配置,是topic和fanout模式下都可以使用的配置。

所以,这里定义了3个queue,2个key(fanout模式下不能使用key),2个key模式匹配(用于匹配topic模式下的key),三个exchange。概念可以会看1.1小节。

三、RabbitMQ的初始化配置

RabbitMQ需要将所有的queue、exchange、binding声明为bean。同时可以定义一个MessageConverter,用于消息的转换。

RabbitConfiguration :

package com.cff.springbootwork.rabbitmq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@EnableRabbit
@Configuration
public class RabbitConfiguration {
	// 队列 分类配置
	@Value("${rabbit.direct.queue}")
	private String directQueueName;

	@Value("${rabbit.common.queue}")
	private String commonQueueName;

	@Value("${rabbit.common.queue2}")
	private String commonQueueName2;

	// key 分类配置
	@Value("${rabbit.direct.key}")
	private String rabbitDirectKey;

	@Value("${rabbit.common.queue.pattern}")
	private String rabbitCommonKeyPattern;

	@Value("${rabbit.common.queue2.pattern}")
	private String rabbitCommonKeyPattern2;

	// exchange 分类配置
	@Value("${rabbit.direct.exchange}")
	private String rabbitDirectExchange;

	@Value("${rabbit.fanout.exchange}")
	private String rabbitFanoutExchange;

	@Value("${rabbit.topic.exchange}")
	private String rabbitTopicExchange;

	// 队列声明
	@Bean(name = "directQueue")
	public Queue directQueue() {
		return new Queue(directQueueName);
	}

	@Bean(name = "commonQueueName")
	public Queue commonQueueName() {
		return new Queue(commonQueueName);
	}

	@Bean(name = "commonQueueName2")
	public Queue commonQueueName2() {
		return new Queue(commonQueueName2);
	}

	// exchange声明
	@Bean(name = "directExchange")
	public DirectExchange directExchange() {
		return new DirectExchange(rabbitDirectExchange);
	}

	@Bean(name = "fanoutExchange")
	public FanoutExchange fanoutExchange() {
		return new FanoutExchange(rabbitFanoutExchange);
	}

	@Bean(name = "topicExchange")
	public TopicExchange topicExchange() {
		return new TopicExchange(rabbitTopicExchange);
	}

	// 数据绑定声明
	@Bean(name = "directBinding")
	public Binding directBinding() {
		return BindingBuilder.bind(directQueue()).to(directExchange()).with(rabbitDirectKey);
	}

	@Bean(name = "fanoutBinding")
	public Binding fanoutBinding() {
		return BindingBuilder.bind(commonQueueName()).to(fanoutExchange());
	}

	@Bean(name = "fanoutBinding2")
	public Binding fanoutBinding2() {
		return BindingBuilder.bind(commonQueueName2()).to(fanoutExchange());
	}

	@Bean(name = "topicBinding1")
	public Binding topicBinding1() {
		return BindingBuilder.bind(commonQueueName()).to(topicExchange()).with(rabbitCommonKeyPattern);
	}

	@Bean(name = "topicBinding2")
	public Binding topicBinding2() {
		return BindingBuilder.bind(commonQueueName2()).to(topicExchange()).with(rabbitCommonKeyPattern2);
	}

	@Bean(name = "topicBinding3")
	public Binding topicBinding3() {
		return BindingBuilder.bind(directQueue()).to(topicExchange()).with(rabbitDirectKey);
	}

	// 数据转换声明
	@Bean
	public MessageConverter myMessageConverter() {
		return new Jackson2JsonMessageConverter();
	}

	@Bean
	public SimpleRabbitListenerContainerFactory myFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
			ConnectionFactory connectionFactory) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		configurer.configure(factory, connectionFactory);
		factory.setMessageConverter(myMessageConverter());
		return factory;
	}
}

这里面:

  • Queue是声明队列,上面声明了三个队列。
  • DirectExchange是Direct模式下的Exchange;
  • FanoutExchange是Fanout模式下的Exchange;
  • TopicExchange是Topic模式下的Exchange;
  • Binding,这里定义了6个绑定,一个绑定到DirectExchange;两个绑定到FanoutExchange;三个绑定到TopicExchange;
  • 声明了一个MessageConverter,用于将实体转换为json,或者将json转换为实体。过程是spring控制自动转换的。
  • 声明一个SimpleRabbitListenerContainerFactory,将MessageConverter传递给SimpleRabbitListenerContainerFactory。

四、配置发送方(生产者)

下面介绍下三种模式是怎么发送数据的。

4.1 Direct模式

PomitDirectMessageService :

package com.cff.springbootwork.rabbitmq.service;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.cff.springbootwork.rabbitmq.dto.ParamReq;

@Service
public class PomitDirectMessageService {
	@Value("${rabbit.direct.key}")
	private String rabbitDirectKey;	
	@Value("${rabbit.direct.exchange}")
	private String rabbitDirectExchange;
	
	@Autowired
	AmqpTemplate amqpTemplate;
	
	public void sendMessage(ParamReq paramReq){
		amqpTemplate.convertAndSend(rabbitDirectExchange,rabbitDirectKey,paramReq);
	}
}

这里的amqpTemplate是Spring自动生成的bean。消息的发送指定了rabbitDirectExchange、指定了rabbitDirectKey。

4.2 Fanout模式

PomitFanoutMessageService :

package com.cff.springbootwork.rabbitmq.service;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.cff.springbootwork.rabbitmq.dto.ParamReq;

@Service
public class PomitFanoutMessageService {
	@Value("${rabbit.fanout.exchange}")
	private String rabbitFanoutExchange;
	@Autowired
	AmqpTemplate amqpTemplate;

	public void sendMessage(ParamReq paramReq) {
		amqpTemplate.convertAndSend(rabbitFanoutExchange, null, paramReq);
	}
}

这里的amqpTemplate是Spring自动生成的bean。消息的发送指定了rabbitFanoutExchange,并没有routingKey.

4.3 Topic模式

PomitTopicMessageService :

package com.cff.springbootwork.rabbitmq.service;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.cff.springbootwork.rabbitmq.dto.ParamReq;

@Service
public class PomitTopicMessageService {
	@Value("${rabbit.topic.exchange}")
	private String rabbitTopicExchange;
	
	@Value("${rabbit.direct.key}")
	private String rabbitDirectKey;
	
	@Value("${rabbit.topic.key}")
	private String rabbitTopicKey;
	
	@Autowired
	AmqpTemplate amqpTemplateTopic;
	
	public void sendMessage(ParamReq paramReq){
		amqpTemplateTopic.convertAndSend(rabbitTopicExchange, rabbitDirectKey, paramReq);
	}
	
	public void sendMessage2(ParamReq paramReq){
		amqpTemplateTopic.convertAndSend(rabbitTopicExchange, rabbitTopicKey, paramReq);
	}
}

这里的amqpTemplate是Spring自动生成的bean。消息的发送指定了rabbitTopicExchange、指定了routingKey,这里发送了两个key,是为了分别测试指定key和模糊匹配key的接收情况。

五、消息接收方(消费者)

我么可以使用@RabbitListener来定义接收方,每一个queue定义一个@RabbitListener即可。因为queue是和模式无关的。参数可以直接传递实体过来,因为Spring根据我们上面定义的MessageConverter会自动将数据转换为实体。

PomitMessageListener:

package com.cff.springbootwork.rabbitmq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import com.cff.springbootwork.rabbitmq.dto.ParamReq;

@Component
public class PomitMessageListener{

	@RabbitListener(queues = "${rabbit.direct.queue}")
	public void handleMessage(ParamReq paramReq) {
		
		System.out.println(paramReq);
	}
	@RabbitListener(queues = "${rabbit.common.queue}")
	public void handleCommonMessage(ParamReq paramReq) {
		
		System.out.println("一、" + paramReq);
	}

	@RabbitListener(queues = "${rabbit.common.queue2}")
	public void handleCommonMessage2(ParamReq paramReq) {
		
		System.out.println("二、" + paramReq);
	}
}

六、过程中用到的实体及测试类

ParamReq :

package com.cff.springbootwork.rabbitmq.dto;

public class ParamReq {
	private String messageType;

	private String content;

	public String getMessageType() {
		return messageType;
	}

	public void setMessageType(String messageType) {
		this.messageType = messageType;
	}

	public String getContent() {
		return content;
	}

	public void setContent(String content) {
		this.content = content;
	}

	@Override
	public String toString() {
		return "ParamReq [messageType=" + messageType + ", content=" + content + "]";
	}

}

RabbitWeb:

package com.cff.springbootwork.rabbitmq.web;

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;

import com.cff.springbootwork.rabbitmq.dto.ParamReq;
import com.cff.springbootwork.rabbitmq.service.PomitDirectMessageService;
import com.cff.springbootwork.rabbitmq.service.PomitFanoutMessageService;
import com.cff.springbootwork.rabbitmq.service.PomitTopicMessageService;

@RestController
@RequestMapping("/rabbit")
public class RabbitWeb {
	@Autowired
	PomitDirectMessageService pomitDirectMessageService;
	
	@Autowired
	PomitFanoutMessageService pomitFanoutMessageService;
	
	@Autowired
	PomitTopicMessageService pomitTopicMessageService;

	@RequestMapping(value = "/direct", method = { RequestMethod.GET })
	public String direct() {
		ParamReq paramReq = new ParamReq();
		paramReq.setContent("asdasdxbcvbcb");
		paramReq.setMessageType("direct");
		pomitDirectMessageService.sendMessage(paramReq);
		return "0000";
	}
	
	@RequestMapping(value = "/fanout", method = { RequestMethod.GET })
	public String fanout() {
		ParamReq paramReq = new ParamReq();
		paramReq.setContent("asdasdxbcvbcb");
		paramReq.setMessageType("fanout");
		pomitFanoutMessageService.sendMessage(paramReq);
		return "0000";
	}
	
	@RequestMapping(value = "/topic1", method = { RequestMethod.GET })
	public String topic1() {
		ParamReq paramReq = new ParamReq();
		paramReq.setContent("asdasdxbcvbcb");
		paramReq.setMessageType("topic1");
		pomitTopicMessageService.sendMessage(paramReq);
		return "0000";
	}
	
	@RequestMapping(value = "/topic2", method = { RequestMethod.GET })
	public String topic2() {
		ParamReq paramReq = new ParamReq();
		paramReq.setContent("asdasdxbcvbcb");
		paramReq.setMessageType("topic2");
		pomitTopicMessageService.sendMessage2(paramReq);
		return "0000";
	}
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • SpringBoot入门建站全系列(十八)整合RabbitMQ(AMQP类消息队列)

    消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间...

    品茗IT
  • Spring和WebSocket整合详解(建立Web聊天室)

    WebSocket 是 HTML5 开始提供的一种在单个 TCP 连接上进行全双工通讯的协议。

    品茗IT
  • SpringBoot入门建站全系列(三十五)整合Oauth2做单机版认证授权

    OAuth 2.0 规范定义了一个授权(delegation)协议,对于使用Web的应用程序和API在网络上传递授权决策非常有用。OAuth被用在各钟各样的应用...

    品茗IT
  • SpringBoot整合RabbitMQ消息队列-学习笔记 原

    本篇文章主要用于记录个人学习RabbitMQ的过程,记录下来方便日后查看,如有错误的地方,还望指正。

    拓荒者
  • SpringBoot入门建站全系列(十八)整合RabbitMQ(AMQP类消息队列)

    消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间...

    品茗IT
  • spring-boot ---Scheduled 定时任务

    用户5927264
  • Spring Boot:整合JdbcTemplate

    Spring对数据库的操作在jdbc上面做了更深层次的封装,而JdbcTemplate便是Spring提供的一个操作数据库的便捷工具。我们可以借助JdbcTem...

    朝雨忆轻尘
  • springmvc 项目完整示例02 项目创建-eclipse创建动态web项目 配置文件 junit单元测试

    spring原理案例-基本项目搭建 01 spring framework 下载 官网下载spring jar包

    noteless
  • Spring和WebSocket整合详解(建立Web聊天室)

    WebSocket 是 HTML5 开始提供的一种在单个 TCP 连接上进行全双工通讯的协议。

    品茗IT
  • 实战:十分钟实现基于JWT前后端分离的权限框架

    面试过很多Java开发,能把权限这块说的清楚的实在是不多,很多人因为公司项目职责问题,很难学到这类相关的流程和技术,本文梳理一个简单的场景,实现一个基于jwt前...

    JAVA葵花宝典

扫码关注云+社区

领取腾讯云代金券