SpringBoot2.0应用:SpringBoot2.0整合RabbitMQ

Posted yjtz

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了SpringBoot2.0应用:SpringBoot2.0整合RabbitMQ相关的知识,希望对你有一定的参考价值。

如何整合RabbitMQ

1、添加spring-boot-starter-amqp

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

2、添加配置

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.dynamic=true
spring.rabbitmq.cache.connection.mode=channel

3、注入队列

@Configuration
public class RabbitConfig {
    @Bean
    public Queue Queue() {
        return new Queue("hello");
    }
}

4、创建操作数据的Repository对象

interface CityRepository extends Repository<City, Long> {

    Page<City> findAll(Pageable pageable);

    Page<City> findByNameContainingAndCountryContainingAllIgnoringCase(String name,
            String country, Pageable pageable);

    City findByNameAndCountryAllIgnoringCase(String name, String country);

}

5、创建消费者

@Component
public class RabbitConsumer {
    @RabbitHandler
    @RabbitListener(queues = "hello")
    public void process(@Payload String foo) {
        System.out.println(new Date() + ": " + foo);
    }
}

6、启动主类

@SpringBootApplication
@EnableScheduling
public class AmqpApplication {
    public static void main(String[] args) {
        SpringApplication.run(AmqpApplication.class, args);
    }
}

控制台输出:

Sun Sep 30 16:30:35 CST 2018: hello

到此,一个简单的SpringBoot2.0集成RabbitMQ就完成了。
熟悉RabbitMQ的小伙伴们应该知道,RabbitMQ在一般的队列基础上,增加了ExChange的概念。ExChange有四种类型:Direct, Topic, Headers and Fanout。其中Headers实际很少使用,Direct较为简单。接下来将详细介绍如何使用topic和Fanout。

Topic Exchange

1、配置Topic规则

@Configuration
public class TopicRabbitConfig {

    @Bean
    public Queue queueMessage1() {
        return new Queue(MQConst.TOPIC_QUEUENAME1);
    }

    @Bean
    public Queue queueMessage2() {
        return new Queue(MQConst.TOPIC_QUEUENAME2);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(MQConst.TOPIC_EXCHANGE);
    }

    @Bean
    Binding bindingExchangeMessage(Queue queueMessage1, TopicExchange exchange) {
        // 将队列1绑定到名为topicKey.A的routingKey
        return BindingBuilder.bind(queueMessage1).to(exchange).with(MQConst.TOPIC_KEY1);
    }

    @Bean
    Binding bindingExchangeMessages(Queue queueMessage2, TopicExchange exchange) {
        // 将队列2绑定到所有topicKey.开头的routingKey
        return BindingBuilder.bind(queueMessage2).to(exchange).with(MQConst.TOPIC_KEYS);
    }
}

2、配置消费者

@Component
public class TopicConsumer {

    @RabbitListener(queues = MQConst.TOPIC_QUEUENAME1)
    @RabbitHandler
    public void process1(String message) {
        System.out.println("queue:topic.message1,message:" + message);
    }

    @RabbitListener(queues = MQConst.TOPIC_QUEUENAME2)
    @RabbitHandler
    public void process2(String message) {
        System.out.println("queue:topic.message2,message:" + message);
    }
}

3、生产消息

在Producer类中添加:

        // Topic
        rabbitTemplate.convertAndSend(MQConst.TOPIC_EXCHANGE, MQConst.TOPIC_KEYS, "from keys");
        rabbitTemplate.convertAndSend(MQConst.TOPIC_EXCHANGE, MQConst.TOPIC_KEY1, "from key1");

再次启动主类,控制台输出:

queue:topic.message2,message:from keys
queue:topic.message1,message:from key1
queue:topic.message2,message:from key1

Fanout Exchange

1、配置Fanout规则

@Configuration
public class FanoutRabbitConfig {
    @Bean
    public Queue MessageA() {
        return new Queue(MQConst.FANOUT_QUEUENAME1);
    }

    @Bean
    public Queue MessageB() {
        return new Queue(MQConst.FANOUT_QUEUENAME2);
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(MQConst.FANOUT_EXCHANGE);
    }

    @Bean
    Binding bindingExchangeA(Queue MessageA, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(MessageA).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue MessageB, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(MessageB).to(fanoutExchange);
    }
}

2.配置消费者

@Component
public class FanoutConsumer {

    @RabbitListener(queues = MQConst.FANOUT_QUEUENAME1)
    @RabbitHandler
    public void process1(String message) {
        System.out.println("queue:fanout.message1,message:" + message);
    }

    @RabbitListener(queues = MQConst.FANOUT_QUEUENAME2)
    @RabbitHandler
    public void process2(String message) {
        System.out.println("queue:fanout.message2,message:" + message);
    }
}

3、生产消息

在Producer类中添加:

        // FanOut
        rabbitTemplate.convertAndSend(MQConst.FANOUT_EXCHANGE, "", "fanout"); 

再次启动主类,控制台输出:

queue:fanout.message2,message:fanout
queue:fanout.message1,message:fanout

源码地址:GitHub


本篇到此结束,如果读完觉得有收获的话,欢迎点赞、关注、加公众号【贰级天災】,查阅更多精彩历史!!!

技术图片

以上是关于SpringBoot2.0应用:SpringBoot2.0整合RabbitMQ的主要内容,如果未能解决你的问题,请参考以下文章

docker-compose启动springboo应用

2018最新SpringBoot2.0教程(零基础入门)

Docker + SpringBoot2.0快速构建镜像

Spring Boot 2.0 使用模板引擎返回JSP页面实战

SpringBoot2.0源码分析:整合ActiveMQ分析

SpringBoot2.0 整合 SpringSecurity 框架,实现用户权限安全管理