消息中间件(十)-----RabbitMq集成springboot

2021-01-19 20:13

阅读：395

标签：factor 优先 cal cse 路由 return end fan report

引入jar包

xml version="1.0" encoding="UTF-8"?>
project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    modelVersion>4.0.0modelVersion>

    groupId>cn.enjoyedugroupId>
    artifactId>demoartifactId>
    version>0.0.1-SNAPSHOTversion>
    packaging>jarpackaging>

    name>rq-springboot-withname>
    description>rq-springboot-withdescription>

    parent>
        groupId>org.springframework.bootgroupId>
        artifactId>spring-boot-starter-parentartifactId>
        version>1.5.8.RELEASEversion>
        relativePath/> 
    parent>

    properties>
        project.build.sourceEncoding>UTF-8project.build.sourceEncoding>
        project.reporting.outputEncoding>UTF-8project.reporting.outputEncoding>
        java.version>1.8java.version>
    properties>

    dependencies>
        dependency>
            groupId>org.springframework.bootgroupId>
            artifactId>spring-boot-starter-amqpartifactId>
        dependency>
        dependency>
            groupId>org.springframework.bootgroupId>
            artifactId>spring-boot-starter-webartifactId>
        dependency>

        dependency>
            groupId>org.springframework.bootgroupId>
            artifactId>spring-boot-starter-testartifactId>
            scope>testscope>
        dependency>
    dependencies>

    build>
        plugins>
            plugin>
                groupId>org.springframework.bootgroupId>
                artifactId>spring-boot-maven-pluginartifactId>
            plugin>
        plugins>
    build>


project>

配置文件

application.properties

spring.application.name=springboot-rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.virtual-host=/

不过这些配置文件优先放在apollo等，方便更改。

RabbitConfig

package cn.enjoyedu.config;

import cn.enjoyedu.RmConst;
import cn.enjoyedu.hello.UserReceiver;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String addresses;

    @Value("${spring.rabbitmq.port}")
    private String port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirms;

    @Autowired
    private UserReceiver userReceiver;

    /**
     * 连接工厂
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {

        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(addresses+":"+port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        /** 如果要进行消息回调，则这里必须要设置为true */
        connectionFactory.setPublisherConfirms(publisherConfirms);
        return connectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        return new RabbitAdmin(connectionFactory);
    }

    /**
     * 自定义rabbitTemplate
     * @return
     */
    @Bean
    public RabbitTemplate newRabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        //设置mandatory参数，发送消息后，成功绑定到队列返回成功给生产者
        template.setMandatory(true);
        //生产者消息确认
        template.setConfirmCallback(confirmCallback());
        //rabbitmq回调(例如设置mandatory参数，回调给生产者)
        template.setReturnCallback(returnCallback());
        return template;
    }

    //===============使用了RabbitMQ系统缺省的交换器==========
    //绑定键即为队列名称
    @Bean
    public Queue helloQueue() {
        return new Queue(RmConst.QUEUE_HELLO);
    }

    @Bean
    public Queue userQueue() {
        return new Queue(RmConst.QUEUE_USER);
    }

    //===============以下是验证topic Exchange==========
    @Bean
    public Queue queueEmailMessage() {
        return new Queue(RmConst.QUEUE_TOPIC_EMAIL);
    }

    @Bean
    public Queue queueUserMessages() {
        return new Queue(RmConst.QUEUE_TOPIC_USER);
    }

    /**
     * 声明topic模式的exchange
     * @return
     */
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(RmConst.EXCHANGE_TOPIC);
    }

    /**
     * 配置exchange、queue以及路由键之间的关系，简单说就是将队列以某个路由键绑定到exchange上
     * @return
     */
    @Bean
    public Binding bindingEmailExchangeMessage() {
        return BindingBuilder
                .bind(queueEmailMessage())
                .to(exchange())
                .with("sb.*.email");
    }

    @Bean
    public Binding bindingUserExchangeMessages() {
        return BindingBuilder
                .bind(queueUserMessages())
                .to(exchange())
                .with("sb.*.user");
    }

    //===============以上是验证topic Exchange==========

    //===============以下是验证Fanout Exchange==========
    @Bean
    public Queue AMessage() {
        return new Queue("sb.fanout.A");
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(RmConst.EXCHANGE_FANOUT);
    }

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

    //===============以上是验证Fanout Exchange的交换器==========

    //===============消费者确认==========
    @Bean
    public SimpleMessageListenerContainer messageContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueues(userQueue());
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(userReceiver);
        return container;
    }

    //===============生产者发送确认==========
    @Bean
    public RabbitTemplate.ConfirmCallback confirmCallback(){
        return new RabbitTemplate.ConfirmCallback(){

            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                if (ack) {
                    System.out.println("发送者确认发送给mq成功");
                } else {
                    //处理失败的消息
                    System.out.println("发送者发送给mq失败,考虑重发:"+cause);
                }
            }
        };
    }

    @Bean
    public RabbitTemplate.ReturnCallback returnCallback(){
        return new RabbitTemplate.ReturnCallback(){

            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("无法路由的消息，需要考虑另外处理。");
                System.out.println("Returned replyText："+replyText);
                System.out.println("Returned exchange："+exchange);
                System.out.println("Returned routingKey："+routingKey);
                String msgJson  = new String(message.getBody());
                System.out.println("Returned Message："+msgJson);
            }
        };
    }
}

生产者

TopicSender

package cn.enjoyedu.topic;

import cn.enjoyedu.RmConst;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class TopicSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send() {
        String msg1 = "I am email mesaage msg======";
        System.out.println("TopicSender send the 1st : " + msg1);
        this.rabbitTemplate.convertAndSend(RmConst.EXCHANGE_TOPIC, RmConst.RK_EMAIL, msg1);

        String msg2 = "I am user mesaages msg########";
        System.out.println("TopicSender send the 2nd : " + msg2);
        this.rabbitTemplate.convertAndSend(RmConst.EXCHANGE_TOPIC, RmConst.RK_USER, msg2);

        String msg3 = "I am error mesaages msg";
        System.out.println("TopicSender send the 3rd : " + msg3);
        this.rabbitTemplate.convertAndSend(RmConst.EXCHANGE_TOPIC, "errorkey", msg3);
    }

}

消费者

TopicEmailMessageReceiver

package cn.enjoyedu.topic;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = "sb.info.email")
public class TopicEmailMessageReceiver {

    @RabbitHandler
    public void process(String msg) {
        System.out.println("TopicEmailMessageReceiver  : " +msg);
    }

}

消息中间件(十)-----RabbitMq集成springboot

标签：factor 优先 cal cse 路由 return end fan report

原文地址：https://www.cnblogs.com/alimayun/p/12906253.html

上一篇：[Java] Java核心（2）深入理解线程池ThreadPool

下一篇：java反射机制的作用与优点

文章来自：搜素材网的编程语言模块，转载请注明文章出处。
文章标题：消息中间件(十)-----RabbitMq集成springboot
文章链接：http://soscw.com/index.php/essay/44210.html

亲，登录后才可以留言！

消息中间件(十)-----RabbitMq集成springboot

引入jar包

配置文件

application.properties

RabbitConfig

生产者

消费者

评论

热门文章

推荐文章

最新文章

置顶文章