0
点赞
收藏
分享

微信扫一扫

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)


摘要:Spring Boot 整合RabbitMQ 实现消息发生和接收。


一:RabbitMQ的介绍:

RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache).
消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下:

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)_消息队列

从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息队列,并且当消息队列收到消息之后,接收消息队列传来的消息,并且给予相应的处理.消息队列常用于分布式系统之间互相信息的传递.
对于RabbitMQ来说,除了这三个基本模块以外,还添加了一个模块,即交换机(Exchange).它使得生产者和消息队列之间产生了隔离,生产者将消息发送给交换机,而交换机则根据调度策略把相应的消息转发给对应的消息队列.那么RabitMQ的工作流程如下所示:

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)_Spring Boot_02

紧接着说一下交换机.交换机的主要作用是接收相应的消息并且绑定到指定的队列.交换机有四种类型,分别为Direct,topic,headers,Fanout.


  Direct是RabbitMQ默认的交换机模式,也是最简单的模式.即创建消息队列的时候,指定一个BindingKey.当发送者发送消息的时候,指定对应的Key.当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中.


  topic转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中.


  headers也是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中.


  Fanout是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略.


二:Spring Boot 整合 RabbitMQ 支持三种模式,分别是:Direct模式(相当于一对一模式,一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收!),

Topic转发模式,Fanout Exchange形式(广播模式)


三:Spring Boot 整合 RabbitMQ 之Direct模式:

SpringBoot整合RabbitMQ非常简单!感觉SpringBoot真的极大简化了开发的搭建环境的时间..这样我们程序员就可以把更多的时间用在业务上了,下面开始搭建环境:

首先创建1个maven工程,这是为了模拟分布式应用系统中,新建两个类,一个发送者(Sender),一个接收者(Receiver) 如下图:

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)_apache_03

紧接着,配置pom.xml文件,注意其中用到了springboot对于AMQP(高级消息队列协议,即面向消息的中间件的设计)

<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">

<groupId>com.micai</groupId>
<artifactId>micai-springboot-rabbitmq-7</artifactId>
<version>1.0-SNAPSHOT</version>
<modelVersion>4.0.0</modelVersion>

<packaging>jar</packaging>

<name>micai-springboot-rabbitmq-7</name>
<url>http://maven.apache.org</url>

<!-- Spring Boot 启动父依赖 -->
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.1.RELEASE</version>
</parent>

<dependencies>
<!-- Spring Boot web依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- Spring Boot AMQP依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>


</project>


紧接着,我们编写发送者相关的代码.首先毫无疑问,要书写启动类:


package com.micai.springboot;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
* Spring Boot应用启动类
*
* @author zhaoxinguo on 2017/8/21.
*/
//@EnableScheduling // 定时任务注解
@SpringBootApplication
public class Application {

public static void main(String [] args){

SpringApplication.run(Application.class,args);

}


}


接着在application.properties中,去编辑和RabbitMQ相关的配置信息,配置信息的代表什么内容根据键就能很直观的看出了.这里端口是5672,不是15672...15672是管理端的端口!


spring.application.name=micai-springboot-rabbitmq-7

# MQ配置
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest


随后,配置Queue(消息队列).那注意由于采用的是Direct模式,需要在配置Queue的时候,指定一个键,使其和交换机绑定.

package com.micai.springboot.mq.config;

import com.micai.springboot.base.BaseConfig;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
* 描述:Topic转发模式
* <p>
* Author: 赵新国
* Date: 2017/11/3 18:51
*/
@Configuration
public class SenderConf extends BaseConfig {

// ---------------------------------------------------- Direct 形式 -------------------------------------------- //

@Bean
public Queue queue() {
return new Queue(QUEUE_KEY);
}

// ---------------------------------------------------- Topic 形式 -------------------------------------------- //

/*@Bean(name = "message")
public Queue queueMessage() {
return new Queue("topic.message");
}

@Bean(name = "messages")
public Queue queueMessages() {
return new Queue("topic.messages");
}

@Bean
public TopicExchange exchange() {
return new TopicExchange("exchange");
}

@Bean
Binding bindingExchangeMessage(@Qualifier("message") Queue queueMessage, TopicExchange exchange) {
return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}

@Bean
Binding bindingExchangeMessages(@Qualifier("messages") Queue queueMessages, TopicExchange exchange) {
return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一个词,#表示零个或多个词
}*/

// -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //

/*@Bean(name="Amessage")
public Queue AMessage() {
return new Queue("fanout.A");
}


@Bean(name="Bmessage")
public Queue BMessage() {
return new Queue("fanout.B");
}

@Bean(name="Cmessage")
public Queue CMessage() {
return new Queue("fanout.C");
}

@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");//配置广播路由器
}

@Bean
Binding bindingExchangeA(@Qualifier("Amessage") Queue AMessage,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(AMessage).to(fanoutExchange);
}

@Bean
Binding bindingExchangeB(@Qualifier("Bmessage") Queue BMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(BMessage).to(fanoutExchange);
}

@Bean
Binding bindingExchangeC(@Qualifier("Cmessage") Queue CMessage, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(CMessage).to(fanoutExchange);
}*/

}



接着就可以发送消息啦!在SpringBoot中,我们使用AmqpTemplate去发送消息!代码如下:

package com.micai.springboot.mq;

import com.micai.springboot.base.BaseConfig;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
* 描述:消息生产者
* <p>
* Author: 赵新国
* Date: 2017/11/3 15:37
*/
@Component
public class Sender extends BaseConfig {

@Autowired
private AmqpTemplate rabbitTemplate;

public void send() {
// ---------------------------------------------------- Direct 形式 -------------------------------------------- //
// 在该生产者,我们会产生一个字符串,并发送到名为hello的队列中
String context = "Hello " + "Rabbit MQ!";
System.out.println("发送MQ消息 : " + context);
this.rabbitTemplate.convertAndSend(QUEUE_KEY, context);

// 发送对象,但是该对象必须实现Serializable接口
/*User user = new User(); //实现Serializable接口
user.setId(1L);
user.setName("张三");
this.rabbitTemplate.convertAndSend(QUEUE_KEY, user);*/

// ---------------------------------------------------- Topic 形式 -------------------------------------------- //
/*this.rabbitTemplate.convertAndSend("exchange", "topic.message", "hello, rabbit!");*/

// -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
/*this.rabbitTemplate.convertAndSend("fanoutExchange", "", "xixi,hlhdidi");// 参数2将被忽略*/
}

}


编写测试类!这样我们的发送端代码就编写完了~

package com.micai;

import com.micai.springboot.Application;
import com.micai.springboot.mq.Sender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
* 描述:
* <p>
* Author: 赵新国
* Date: 2017/11/3 17:21
*/
@SpringBootTest(classes=Application.class)
@RunWith(SpringJUnit4ClassRunner.class)
public class RabbitMQTest {

@Autowired
private Sender sender;

@Test
public void send (){
sender.send();
}

}



接着我们编写接收端.接收端的pom文件,application.properties(修改spring.application.name),Queue配置类,App启动类都是一致的!这里省略不计.主要在于我们需要配置监听器去监听绑定到的消息队列,当消息队列有消息的时候,予以接收,代码如下:

package com.micai.springboot.mq;

import com.micai.springboot.base.BaseConfig;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
* 描述:消息消费者
* @RabbitListener注解定义该类对hello队列的监听,
* 并用@RabbitHandler注解来指定对消息的处理方法。
* 所以,该消费者实现了对hello队列的消费,消费操作为输出消息的字符串内容
* Author: 赵新国
* Date: 2017/11/3 15:42
*/
@Component
public class Receiver extends BaseConfig {

// ---------------------------------------------------- Direct 形式 -------------------------------------------- //
//监听器监听指定的Queue
@RabbitListener(queues = QUEUE_KEY)
public void process(String str) {
System.out.println("接收MQ消息 : " + str);
}

// //监听器监听指定的Queue
/*@RabbitListener(queues = QUEUE_KEY)
public void process(User user) { //用User作为参数
System.out.println("接收MQ消息 : " + user);
}

// ---------------------------------------------------- Topic 形式 -------------------------------------------- //
@RabbitListener(queues="topic.message") //监听器监听指定的Queue
public void process1(String str) {
System.out.println("message:"+str);
}

@RabbitListener(queues="topic.messages") //监听器监听指定的Queue
public void process2(String str) {
System.out.println("messages:"+str);
}*/


// -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
/*@RabbitListener(queues="fanout.A")
public void processA(String str1) {
System.out.println("ReceiveA:"+str1);
}

@RabbitListener(queues="fanout.B")
public void processB(String str) {
System.out.println("ReceiveB:"+str);
}

@RabbitListener(queues="fanout.C")
public void processC(String str) {
System.out.println("ReceiveC:"+str);
}*/

}


接下来就可以测试啦,首先启动接收端的应用,紧接着运行发送端的单元测试,接收端应用打印出来接收到的消息,测试即成功!



需要注意的地方,Direct模式相当于一对一模式,一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收!

运行结果如下图:

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)_apache_04

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)_Spring Boot_05


实际上RabbitMQ还可以支持发送对象:当然由于涉及到序列化和反序列化,该对象要实现Serilizable接口.Sender做出如下改写:

package com.micai.springboot.mq;

import com.micai.springboot.base.BaseConfig;
import com.micai.springboot.entity.User;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
* 描述:消息生产者
* <p>
* Author: 赵新国
* Date: 2017/11/3 15:37
*/
@Component
public class Sender extends BaseConfig {

@Autowired
private AmqpTemplate rabbitTemplate;

public void send() {
// ---------------------------------------------------- Direct 形式 -------------------------------------------- //
// 在该生产者,我们会产生一个字符串,并发送到名为hello的队列中
/*String context = "Hello " + "Rabbit MQ!";
System.out.println("发送MQ消息 : " + context);
this.rabbitTemplate.convertAndSend(QUEUE_KEY, context);*/

// 发送对象,但是该对象必须实现Serializable接口
User user = new User(); //实现Serializable接口
user.setId(1L);
user.setName("张三");
this.rabbitTemplate.convertAndSend(QUEUE_KEY, user);

// ---------------------------------------------------- Topic 形式 -------------------------------------------- //
/*this.rabbitTemplate.convertAndSend("exchange", "topic.message", "hello, rabbit!");*/

// -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
/*this.rabbitTemplate.convertAndSend("fanoutExchange", "", "xixi,hlhdidi");// 参数2将被忽略*/
}

}


Receiver做出如下改写:

package com.micai.springboot.mq;

import com.micai.springboot.base.BaseConfig;
import com.micai.springboot.entity.User;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
* 描述:消息消费者
* @RabbitListener注解定义该类对hello队列的监听,
* 并用@RabbitHandler注解来指定对消息的处理方法。
* 所以,该消费者实现了对hello队列的消费,消费操作为输出消息的字符串内容
* Author: 赵新国
* Date: 2017/11/3 15:42
*/
@Component
public class Receiver extends BaseConfig {

// ---------------------------------------------------- Direct 形式 -------------------------------------------- //
//监听器监听指定的Queue
/*@RabbitListener(queues = QUEUE_KEY)
public void process(String str) {
System.out.println("接收MQ消息 : " + str);
}*/

// //监听器监听指定的Queue
@RabbitListener(queues = QUEUE_KEY)
public void process(User user) { //用User作为参数
System.out.println("接收MQ消息 : " + user);
}

// ---------------------------------------------------- Topic 形式 -------------------------------------------- //
/*@RabbitListener(queues="topic.message") //监听器监听指定的Queue
public void process1(String str) {
System.out.println("message:"+str);
}

@RabbitListener(queues="topic.messages") //监听器监听指定的Queue
public void process2(String str) {
System.out.println("messages:"+str);
}*/


// -------------------------------------------- Fanout Exchange形式 ------------------------------------------- //
/*@RabbitListener(queues="fanout.A")
public void processA(String str1) {
System.out.println("ReceiveA:"+str1);
}

@RabbitListener(queues="fanout.B")
public void processB(String str) {
System.out.println("ReceiveB:"+str);
}

@RabbitListener(queues="fanout.C")
public void processC(String str) {
System.out.println("ReceiveC:"+str);
}*/

}


改写后的运行结果如下:

Spring Boot 整合 RabbitMQ 之 Direct模式 (一)_RabbitMQ_06




举报

相关推荐

0 条评论