后端技术

spring boot集成消息中间件（rabbitMQ）

2018-11-23  本文已影响48人  Mysql_rong

前言

目前市面上应用比较多的几类消息中间件主要有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 消息处理完成
上一篇下一篇

猜你喜欢

热点阅读