Day10_06_消息队列之SpringBoot结合RabbitMQ使用教程

SpirngBoot结合RabbitMQ使用教程 一. 创建项目

创建一个新的项目,整个项目目录结构如下

1. 父项目的pom.xml文件

    
    4.0.0
    com.syc.boot
    Day12_boot_amqp
    pom
    1.0-SNAPSHOT

    
        amqp-producer
        amqp-consumer
        amqp-commons
    

    
        org.springframework.boot
        spring-boot-starter-parent
        1.5.12.RELEASE
         
    

    
        UTF-8
        UTF-8
        UTF-8
        1.8
        1.8
        1.8

        5.0.7
        1.3.1
        1.16.20
    

    
        
        
            org.springframework.boot
            spring-boot-starter-web
        

        
        
            org.springframework.boot
            spring-boot-starter-amqp
        

        
        
            org.projectlombok
            lombok
            ${lombok.version}
        

        
            org.springframework.boot
            spring-boot-starter-test
            test
        
    

    
        
            
                
                    org.springframework.boot
                    spring-boot-maven-plugin
                    ${spring-boot.version}
                
                
                    maven-compiler-plugin
                    
                        ${java.version}
                        ${java.version}
                        UTF-8
                    
                
            
        
    

二. 创建消息生产者模块 1. 创建项目模块

创建一个名为 amqp-producer 的消息生产者模块. 该模块的目录结构为:

2. application.yml

spring:
  application:
    name: amqp-producer
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

3. 创建队列配置

package com.syc.boot.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 4种交换机之一:
 * 直接型交换机,也是默认的交换机.
 */
@Configuration
public class DirectRabbitMQConfig {

    //交换机和队列要进行绑定:
    //默认的交换机是DirectExchange,
    //每个交换机都需要利用路由键来和队列绑定在一起.
    //如果采用的是DirectExchange交换机,默认情况下,队里的名字就是路由键的名字.
    //该交换机是一对一的,一个消息被发送者发送出去之后,只能被一个消费者接受.
    @Bean
    public Queue queue(){
        //firstRabbit,是队列的名字,
        return new Queue("firstRabbit");
    }

    @Bean
    public Queue objectQueue(){
        //firstRabbit,是队列的名字,
        return new Queue("firstObject");
    }
}

3. 创建消息提供者

package com.syc.boot;

import com.syc.boot.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 *
 */
@Slf4j
@Component
public class FirstRabbitProducer {

    @Autowired
    private AmqpTemplate template;

    public void sendMsg() {
        //Message msg=new Message();
        //template.send(msg);

        //第一个参数是路由键
        String content="hello,"+new Date();
        log.warn("消息发送--->content={}",content);
        template.convertAndSend("firstRabbit", content);
    }

    //发送对象
    public void sendUser() {
        User user=new User();
        user.setId("1");
        user.setUsername("syc");
        user.setPassword("123");
        log.warn("消息发送--->User={}",user.toString());
        template.convertAndSend("firstObject", user);
    }

}

4. 创建测试用例

package com.syc.boot;

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.SpringRunner;

/**
 *
 */
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ProviderApplication.class)
public class FirstRabbitTest {

    @Autowired
    private FirstRabbitProducer producer;

    @Test
    public void firstSendTest() {
        for (int i = 0; i < 10; i++) {
            producer.sendMsg();
        }
    }

 }

5. 项目入口类

package com.syc.boot;

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

@SpringBootApplication
public class ProviderApplication {

    public static void main(String[] args) {
        SpringApplication.run(ProviderApplication.class, args);
    }
}

6. 启动测试类

执行测试类中的firstSendTest()测试方法.看到控制台打印如下日志.

并且此时RabbitMQ中的firstRabbit队列中有10条待消费的消息.

三. 创建消息消费者模块 1. 创建消费者模块

创建一个名为 amqp-consumer 的消费者模块. 项目结构图:

2. application.yml

spring:
  application:
    name: amqp-consumer
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

3. 创建消息消费者

package com.syc.boot.msg;

import com.syc.boot.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 *
 */
@Slf4j
@Component
public class FirstRabbitConsumer {

    @RabbitListener(queues = "firstRabbit")
    @RabbitHandler
    public void receiveMsg(String msg) {
        log.warn("接受消息---->content:{}", msg);
    }
    
}

4. 项目入口类

package com.syc.boot;

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

@SpringBootApplication
public class ConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }
}

5. 启动项目,接收消息

项目启动后,可以看到控制台打印如下消息

同时RabbitMQ控制台中firstRabbit队列中的消息全部消费掉,消息数量为0.