前言
目前市面上应用比较多的几类消息中间件主要有activeMQ、kafka、rocketMQ、rabbitMQ，大致了解后决定用rabbitMQ作为实践对象，没什么具体原因，大家可以根据自身业务场景及特点针对性地去选择，废话不多说，直接上代码了。
安装rabbitMQ
本人的服务器版本是ubuntu，所以安装rabbitMQ比较简单：
#apt-get -install rabbitmq
启动rabbitMQ
/sbin/service rabbitmq-server start
创建一个测试用户用于远程访问，（默认提供的用户只能本地访问）
# /sbin/rabbitmqctl add_user rabbit 123456
rabbit是我的用户名 123456是密码
开启rabbitMQ的web插件，也就是类似于tomcat、weblogic这种中间件的管理控制台，能通过15672端口进行访问
# rabbitmq-plugins enable rabbitmq_management
确定安装的rabbitMQ能正常访问后，新建一个spring boot项目，勾选上rabbitMQ的starter，我在这用的工具是idea，新建好的项目依赖环境如下：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
创建消息生产者
在application.properties添加一下配置（rabbit服务的默认端口是5672,15672是控制台的端口号）
spring.rabbitmq.host=39.105.196.148
spring.rabbitmq.port=5672
spring.rabbitmq.username=rabbit
spring.rabbitmq.password=123456
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
创建spring boot配置类，主要有指定队列，交换器类型及绑定操作
package com.example.rabbitmq.controller;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
public Queue queue1(){
return new Queue("hello.queue1",true);
}
@Bean
public Queue queue2(){
return new Queue("hello.queue2",true);
}
@Bean
public TopicExchange topicExchange(){
return new TopicExchange("topicExchange");
}
@Bean
public Binding binding1(){
return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");
}
@Bean
public Binding binding2(){
return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");
}
}
以上声明了两个队列queue1 queue2 ，一个交换器，交换器的类型为Topic，除了Topic还有Direct exchange（直连交换机）、Fanout exchange（扇型交换机）、Headers exchange（头交换机）
创建生产者类
package com.example.rabbitmq.controller;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.UUID;
@Component
public class Sender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
@Autowired
RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
System.out.println("消息发送成功：" + correlationData);
} else {
System.out.println("消息发送失败：" + cause);
}
}
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
System.out.println(message.getMessageProperties().getCorrelationId() + " 发送失败");
}
public void send(String msg) {
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
System.out.println("开始发送消息 : " + msg.toLowerCase());
String response=rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();
System.out.println("结束发送消息 : " + msg.toLowerCase());
System.out.println("消费者响应 : " + response + " 消息处理完成");
}
创建消费者类
package com.example.rabbitmq.controller;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class Recevier {
@RabbitListener(queues = "hello.queue1")
public String processMsg1(String msg){
System.out.println(Thread.currentThread().getName()+"接收到来自hello.queue1队列的消息："+msg);
return msg.toUpperCase();
}
@RabbitListener(queues = "hello.queue2")
public void processMessage2(String msg) {
System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue2队列的消息：" + msg);
}
}
定义了2个队列，所以分别定义不同的监听器监听不同的队列。但是队列2绑定交换机的时候key值为key.#,#的意思是有多个，所以在此可以监听到send方法中的key.1的。
创建测试类进行测试
package com.example.rabbitmq;
import com.example.rabbitmq.controller.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;
import java.util.Date;
@RunWith(value = SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = RabbitmqApplication.class)
public class RabbitTest {
@Autowired
Sender sender;
@Test
public void sendTest()throws Exception{
while (true){
String msg = new Date().toString();
sender.send(msg);
Thread.sleep(1000);
}
}
}
下面为输出内容
开始发送消息 : fri nov 23 10:49:36 cst 2018
2018-11-23 10:49:36.541 INFO 3780 --- [ main] o.s.s.c.ThreadPoolTaskScheduler : Initializing ExecutorService
2018-11-23 10:49:36.546 INFO 3780 --- [ main] .l.DirectReplyToMessageListenerContainer : Container initialized for queues: [amq.rabbitmq.reply-to]
2018-11-23 10:49:36.591 INFO 3780 --- [ main] .l.DirectReplyToMessageListenerContainer : SimpleConsumer [queue=amq.rabbitmq.reply-to, consumerTag=amq.ctag-2fRyK459oNmluS71at5PmQ identity=31dfc6f5] started
消息发送成功：CorrelationData [id=1867107d-da6a-4f40-9e6a-eee1e0c8e27b]
SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Fri Nov 23 10:49:36 CST 2018
SimpleAsyncTaskExecutor-1接收到来自hello.queue1队列的消息：Fri Nov 23 10:49:36 CST 2018
结束发送消息 : fri nov 23 10:49:36 cst 2018
消费者响应 : FRI NOV 23 10:49:36 CST 2018 消息处理完成
开始发送消息 : fri nov 23 10:49:37 cst 2018
SimpleAsyncTaskExecutor-1接收到来自hello.queue1队列的消息：Fri Nov 23 10:49:37 CST 2018
SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Fri Nov 23 10:49:37 CST 2018
消息发送成功：CorrelationData [id=b6487669-30dd-4fca-94db-239ee9660466]
结束发送消息 : fri nov 23 10:49:37 cst 2018
消费者响应 : FRI NOV 23 10:49:37 CST 2018 消息处理完成
开始发送消息 : fri nov 23 10:49:38 cst 2018
消息发送成功：CorrelationData [id=d1c023e4-453b-4ccf-94c0-a381921c48ff]
SimpleAsyncTaskExecutor-1接收到来自hello.queue1队列的消息：Fri Nov 23 10:49:38 CST 2018
SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Fri Nov 23 10:49:38 CST 2018
结束发送消息 : fri nov 23 10:49:38 cst 2018
消费者响应 : FRI NOV 23 10:49:38 CST 2018 消息处理完成
网友评论