springBoot+RabbitMQ例子

Posted 2020-09-20 随笔

demo目录

贴代码

1.ProducerConfig.java

package com.test.config;

import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created by admin on 2017/6/1 13:23.
 */
@Configuration
public class ProducerConfig {
    @Bean
    public RabbitMessagingTemplate msgMessageTemplate(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        //参数列表分别是：1.交换器名称(default.topic 为默认值),2.是否长期有效,3.如果服务器在不再使用时自动删除交换器
        TopicExchange exchange = new TopicExchange("default.topic", true, false);
        rabbitAdmin.declareExchange(exchange);
        //1.队列名称,2.声明一个持久队列,3.声明一个独立队列,4.如果服务器在不再使用时自动删除队列
        Queue queue = new Queue("test.demo.send", true, false, false);
        rabbitAdmin.declareQueue(queue);
        //1.queue:绑定的队列,2.exchange:绑定到那个交换器,3.test2.send:绑定的路由名称
        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("test2.send"));
        return RabbitUtil.simpleMessageTemplate(connectionFactory);
    }
}

2.RabbitMQConfig.java

package com.test.config;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created by admin on 2017/6/1 11:26.
 */
@Configuration
public class RabbitMQConfig {
    /**
     * 注入配置文件属性
     */
    @Value("${spring.rabbitmq.addresses}")
    String addresses;//MQ地址
    @Value("${spring.rabbitmq.username}")
    String username;//MQ登录名
    @Value("${spring.rabbitmq.password}")
    String password;//MQ登录密码
    @Value("${spring.rabbitmq.virtual-host}")
    String vHost;//MQ的虚拟主机名


    /**
     * 创建 ConnectionFactory
     *
     * @return
     * @throws Exception
     */
    @Bean
    public ConnectionFactory connectionFactory() throws Exception {
        return RabbitUtil.connectionFactory(addresses, username, password, vHost);
    }

    /**
     * 创建 RabbitAdmin
     *
     * @param connectionFactory
     * @return
     * @throws Exception
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) throws Exception {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        return rabbitAdmin;
    }


}

3.RabbitUtil.java

package com.test.config;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.messaging.converter.GenericMessageConverter;

/**
 * RabbitMQ 公共类
 * Created by admin on 2017/6/1 11:25.
 */
public class RabbitUtil {

    /**
     * 初始化 ConnectionFactory
     *
     * @param addresses
     * @param username
     * @param password
     * @param vHost
     * @return
     * @throws Exception
     */
    public static ConnectionFactory connectionFactory(String addresses, String username, String password, String vHost) throws Exception {
        CachingConnectionFactory factoryBean = new CachingConnectionFactory();
        factoryBean.setVirtualHost(vHost);
        factoryBean.setAddresses(addresses);
        factoryBean.setUsername(username);
        factoryBean.setPassword(password);
        return factoryBean;
    }

    /**
     * 初始化 RabbitMessagingTemplate
     *
     * @param connectionFactory
     * @return
     */
    public static RabbitMessagingTemplate simpleMessageTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        RabbitMessagingTemplate rabbitMessagingTemplate = new RabbitMessagingTemplate();
        rabbitMessagingTemplate.setMessageConverter(new GenericMessageConverter());
        rabbitMessagingTemplate.setRabbitTemplate(template);
        return rabbitMessagingTemplate;
    }
}

4.Student.java

package com.test.model;

import java.io.Serializable;

/**
 * Created by admin on 2017/6/1 13:36.
 */
public class Student implements Serializable {
    private String name;
    private Integer age;
    private String address;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }
}

5.Consumers.java

package com.test.task;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

/**
 * Created by admin on 2017/6/1 13:29.
 */
@Service
public class Consumers {

    @RabbitListener(
            //1.rabbitAdmin:RabbitAdmin名称
            admin = "rabbitAdmin",
            bindings = @QueueBinding(
                    //1.test.demo.send:队列名,2.true:是否长期有效,3.false:是否自动删除
                    value = @Queue(value = "test.demo.send", durable = "true", autoDelete = "false"),
                    //1.default.topic交换器名称(默认值),2.true:是否长期有效,3.topic:类型是topic
                    exchange = @Exchange(value = "default.topic", durable = "true", type = "topic"),
                    //test2.send:路由的名称,ProducerConfig 里面 绑定的路由名称(xxxx.to(exchange).with("test2.send"))) 
                    key = "test2.send")
    )
    public void test(Object obj) {
        System.out.println("receive....");
        System.out.println("obj:" + obj.toString());
    }
}

6.Producers.java

package com.test.task;

import com.test.model.Student;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * Created by admin on 2017/6/1 13:35.
 */
@Service
public class Producers {

    @Autowired
    RabbitMessagingTemplate rabbitSendTemplate;

    public void send(Student student) {
        System.out.println("send start.....");
        rabbitSendTemplate.convertAndSend("default.topic", "test2.send", student);
    }
}

7.TestController.java

package com.test.test;

import com.test.model.Student;
import com.test.task.Producers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

/**
 * Created by admin on 2017/6/1 13:38.
 */
@Controller
@RequestMapping(value = "/test")
public class TestController {

    @Autowired
    Producers producers;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    @ResponseBody
    public void test() {
        Student s = new Student();
        s.setName("zhangsan");
        s.setAddress("wuhan");
        s.setAge(20);
        producers.send(s);
    }


}

8.MainApplication.java

package com.test;

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

/**
 * Created by admin on 2017/6/1 11:19.
 */
@SpringBootApplication
public class MainApplication {
    public static void main(String[] args) {
        System.getProperties().put("test", "test");
        SpringApplication.run(MainApplication.class, args);

    }
}

9.application.yml

server:
    address: 192.168.200.117 #自己主机的IP地址
    port: 8000 #端口
spring:
  rabbitmq:
    addresses: 192.168.200.119:5672 #MQ IP 和 端口
    username: admin #MQ登录名
    password: 123456 #MQ登录密码
    virtual-host: test #MQ的虚拟主机名称

10.build.gradle

group \'rabbitmqtest\'
version \'1.0-SNAPSHOT\'

apply plugin: \'java\'

sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
    testCompile group: \'junit\', name: \'junit\', version: \'4.11\'
    testCompile("org.springframework.boot:spring-boot-starter-test:1.3.5.RELEASE")
    compile("org.springframework.boot:spring-boot-starter-web:1.3.5.RELEASE")
    compile(group: \'org.springframework.amqp\', name: \'spring-rabbit\', version: "1.6.1.RELEASE")
}

11.settings.gradle

rootProject.name = \'rabbitmqtest\'

页面访问 192.168.200.117:8000/test/send  可以看到控制台有日志输出，发送的消息立即消费掉了

MQ的队列里面也是空的

如果把消费者的代码注掉，再访问刚才的 url 地址 队列里面就会多一条