ITPub博客

首页 > 架构设计 > 软件结构 > Java SpringBoot 整合 RabbitMQ

Java SpringBoot 整合 RabbitMQ

原创 软件结构 作者:gung123 时间:2020-02-26 18:37:37 0 删除 编辑

如果要进行 RabbitMQ 整合的时候一定要注意以下几个概念:交换空间、虚拟主机、队列信息。本次为了方便起见将项目分为 两个:RabbitMQ-Consumer、RabbitMQ-Producer。

了解springcloud架构可以加求求:三五三六二四七二五九

1、 【两个项目】将 rabbitmq 的依赖支持包拷贝到项目之中;

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

2、【microboot-rabbitmq-producer、microboot-rabbitmq-consumer】修改 application.yml 配置文件,追加 rabbitmq 的相关配置项:

server:
  port: 80
spring:
  messages:
    basename: i18n/Messages,i18n/Pages
  rabbitmq:
    addresses: rabbitmq-server
    username: studyjava
    password: hello
    virtual-host: /

 3、【microboot-rabbitmq-producer】建立一个消息的发送接口:

package cn.study.microboot.producer;
public interface IMessageProducerService {    
public void sendMessage(String msg) ;
}

4、 【microboot-rabbitmq-producer】为了可以正常使用 RabbitMQ 进行消息处理,你还需要做一个消息生产配置类;

package cn.study.microboot.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ProducerConfig {
    public static final String EXCHANGE = "study.microboot.exchange"; // 交换空间名称
    public static final String ROUTINGKEY = "study.microboot.routingkey"; // 设置路由key
    public static final String QUEUE_NAME = "study.microboot.queue"; // 队列名称
    @Bean
    public Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ;
    }
    @Bean
    public DirectExchange getDirectExchange() { // 使用直连的模式
        return new DirectExchange(EXCHANGE, true, true);
    }
    @Bean
    public Queue queue() { // 要创建的队列信息
        return new Queue(QUEUE_NAME);
    }
}

5、 【microboot-rabbitmq-producer】创建消息服务的实现子类:

package cn.study.microboot.producer.impl;
import javax.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import cn.study.microboot.config.ProducerConfig;
import cn.study.microboot.producer.IMessageProducerService;
@Service
public class MessageProducerServiceImpl implements IMessageProducerService {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Override
    public void sendMessage(String msg) {
        this.rabbitTemplate.convertAndSend(ProducerConfig.EXCHANGE,
                ProducerConfig.ROUTINGKEY, msg);
    }
}

6、 【microboot-rabbitmq-consumer】依然需要做一个消费者的配置程序类,而这个程序类里面主要的目的依然是设置交换空间、 路由 KEY 等信息。

package cn.study.microboot.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ConsumerConfig {
    public static final String EXCHANGE = "study.microboot.exchange"; // 交换空间名称
    public static final String ROUTINGKEY = "study.microboot.routingkey"; // 设置路由key
    public static final String QUEUE_NAME = "study.microboot.queue"; // 队列名称
    @Bean
    public Queue queue() { // 要创建的队列信息
        return new Queue(QUEUE_NAME);
    }
    @Bean
    public DirectExchange getDirectExchange() { // 使用直连的模式
        return new DirectExchange(EXCHANGE, true, true);
    }
    @Bean
    public Binding bindingExchangeQueue(DirectExchange exchange,Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY) ;
    }
}

7、 【microboot-rabbitmq-consumer】实现监听处理类:

package cn.study.microboot.consumer;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
public class MessageConsumerService {
    @RabbitListener(queues="study.microboot.queue")
    public void receiveMessage(String text) {    // 进行消息接收处理
        System.err.println("【*** 接收消息 ***】" + text);
    }
}

8、 【microboot-rabbitmq-producer】创建一个测试类实现消息的发送处理。

package cn.study.microboot.test;
import javax.annotation.Resource;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
import cn.study.microboot.StartSpringBootMain;
import cn.study.microboot.producer.IMessageProducerService;
@SpringBootTest(classes = StartSpringBootMain.class)
@RunWith(SpringJUnit4ClassRunner.class)
@WebAppConfiguration
public class TestActiveMQ {
    @Resource
    private IMessageProducerService messageProducer;
    @Test
    public void testSend() throws Exception {
        for (int x = 0; x < 100; x++) {
            this.messageProducer.sendMessage("study - " + x);
        }
    }
}

9、 【microboot-rabbitmq-consumer】编写消息接收测试类,这里面不需要编写代码,只需要做一个休眠即可:

package cn.study.microboot;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
@SpringBootTest(classes = StartSpringBootMain.class)
@RunWith(SpringJUnit4ClassRunner.class)
@WebAppConfiguration
public class AppTest {
    @Test
    public void testStart() throws Exception {
        Thread.sleep(Long.MAX_VALUE);
    }
}

整体进行项目开发之中整合的处理步骤还是简单,但是千万要注意,由于是第一次整合处理,所以将生产者与消费者的配置 类分开了,实际上这两个类的作用是完全一样的。

来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/69952307/viewspace-2677364/,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2019-11-01

  • 博文量
    99
  • 访问量
    40059