RabbitMQ：Spring + RabbitMQ整合发送同步消息

朴灿烈づ我的快乐病毒、 2022-05-18 00:41 279阅读 0赞

> # **RabbitMQ：Spring + RabbitMQ整合发送同步消息** #

## 注：[RabbitMQ系列代码实例github地址][RabbitMQ_github] ##

### （一）：XML配置方式   ###

**以下实现使用Exchange类型为DirectExchange****【模板默认转发器】****. routingkey****【模板默认选择键】****的名称默认为Queue的名称，运行生产者向队列中发送一条消息，再运行消费者消费消息。**

**注意：事先要在RabbitMQ中增添QUEUE\_NAME 队列名称**

** 1.【maven依赖jar包】：**

<properties>
                <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
                <spring.version>4.1.0.RELEASE</spring.version>
                <spring.amqp.version>1.1.1.RELEASE</spring.amqp.version>
                <mybatis.version>3.2.4</mybatis.version>
            </properties>
    
            <dependencies>
                <dependency>
                    <groupId>junit</groupId>
                    <artifactId>junit</artifactId>
                    <version>4.10</version>
                </dependency>
                <dependency>
                    <groupId>com.dangdang</groupId>
                    <artifactId>sharding-jdbc-core</artifactId>
                    <version>1.0.0</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-orm</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>commons-dbcp</groupId>
                    <artifactId>commons-dbcp</artifactId>
                    <version>1.4</version>
                </dependency>
                <dependency>
                    <groupId>org.mybatis</groupId>
                    <artifactId>mybatis-spring</artifactId>
                    <version>1.2.2</version>
                </dependency>
                <dependency>
                    <groupId>org.mybatis</groupId>
                    <artifactId>mybatis</artifactId>
                    <version>${mybatis.version}</version>
                </dependency>
    
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-expression</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-aop</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-beans</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-context</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-context-support</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-test</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-tx</artifactId>
                    <version>${spring.version}</version>
                </dependency>
                <dependency>
                    <groupId>mysql</groupId>
                    <artifactId>mysql-connector-java</artifactId>
                    <version>5.1.28</version>
                </dependency>
                <dependency>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                    <version>1.2.16</version>
                </dependency>
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                    <version>1.7.5</version>
                </dependency>
                
                <dependency>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>transport</artifactId>
                    <version>5.6.3</version>
                </dependency>
                
                <dependency>
                    <groupId>com.google.code.gson</groupId>
                    <artifactId>gson</artifactId>
                    <version>2.8.0</version>
                </dependency>
                
                <dependency>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-core</artifactId>
                    <version>2.8.2</version>
                </dependency>
                <dependency>
                    <groupId>com.rabbitmq</groupId>
                    <artifactId>amqp-client</artifactId>
                    <version>3.6.5</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework.amqp</groupId>
                    <artifactId>spring-rabbit</artifactId>
                    <version>1.7.5.RELEASE</version>
                </dependency>
                <dependency>
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-core</artifactId>
                    <version>${spring.version}</version>
                    <exclusions>
                        
                        <exclusion>
                            <groupId>commons-logging</groupId>
                            <artifactId>commons-logging</artifactId>
                        </exclusion>
                    </exclusions>
                </dependency>
                <dependency>
                    <groupId>org.springframework.amqp</groupId>
                    <artifactId>spring-amqp</artifactId>
                    <version>${spring.amqp.version}</version>
                    <classifier>sources</classifier>
                    <scope>compile</scope>
                </dependency>
                <dependency>
                    <groupId>commons-lang</groupId>
                    <artifactId>commons-lang</artifactId>
                    <version>2.6</version>
                </dependency>
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-api</artifactId>
                    <version>1.5.10</version>
                </dependency>
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>jcl-over-slf4j</artifactId>
                    <version>1.5.10</version>
                    <scope>runtime</scope>
                </dependency>
                <dependency>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                    <version>1.5.10</version>
                    <scope>runtime</scope>
                </dependency>
                <dependency>
                    <groupId>log4j</groupId>
                    <artifactId>log4j</artifactId>
                    <version>1.2.14</version>
                    <scope>runtime</scope>
                </dependency>
                <dependency>
                    <groupId>org.aspectj</groupId>
                    <artifactId>aspectjweaver</artifactId>
                    <version>1.6.9</version>
                </dependency>
            </dependencies>

**2. 【配置文件】：**

**application.properties：**

#============== rabbitmq config ====================
    rabbit.hosts=127.0.0.1
    rabbit.username=
    rabbit.password=
    rabbit.virtualHost=/
    rabbit.queue=spring.queue.sync
    rabbit.routingKey=spring.queue.sync

**applicationContext.xml：**

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
           xmlns:aop="http://www.springframework.org/schema/aop"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
    						http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
    					    http://www.springframework.org/schema/aop
    	                    http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
    						http://www.springframework.org/schema/context
    						http://www.springframework.org/schema/context/spring-context-3.0.xsd"
           default-lazy-init="false">
    
        <description>Spring公共配置 </description>
        <context:component-scan base-package="com.caox.rabbitmq.demo" />
        <aop:aspectj-autoproxy />
        
        <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
            <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE" />
            <property name="ignoreResourceNotFound" value="true" />
            <property name="locations">
                <list>
                    
                    <value>classpath*:application.properties</value>
                </list>
            </property>
        </bean>
        <import resource="applicationContext-rabbitmq-sync.xml"/>
    </beans>

**applicationContext-rabbitmq-sync.xml :**

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:context="http://www.springframework.org/schema/context"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="
    		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
    		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">
    
    
        
        <bean id="rabbitConnectionFactory"
              class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
            <constructor-arg value="${rabbit.hosts}"/>
            <property name="username" value="${rabbit.username}"/>
            <property name="password" value="${rabbit.password}"/>
            <property name="virtualHost" value="${rabbit.virtualHost}"/>
            <property name="channelCacheSize" value="5"/>
        </bean>
    
        
        
        
        
        
    
    
        
        <bean id="rabbitTemplate"
              class="org.springframework.amqp.rabbit.core.RabbitTemplate">
            <constructor-arg ref="rabbitConnectionFactory"></constructor-arg>
            <property name="queue" value="${rabbit.queue}"></property>
            <property name="routingKey" value="${rabbit.routingKey}"></property>
        </bean>
    
        
        
        
        
        
    </beans>

**3.【生产者和消费者】： **

**生产者Producer：**

package com.caox.rabbitmq.demo._08_spring_rabbitmq_sync;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    /**
     * Created by nazi on 2018/7/27.
     */
    public class Producer {
        public static void main(String[] args) {
            ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
            AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
            amqpTemplate.convertAndSend("test spring sync");
        }
    }

**消费者Consumer： **

package com.caox.rabbitmq.demo._08_spring_rabbitmq_sync;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    /**
     * Created by nazi on 2018/7/27.
     * 消费者
     */
    public class Consumer {
    
        public static void main(String[] args) {
            ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
            AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
            System.out.println("Received: " + amqpTemplate.receiveAndConvert());
        }
    }

### **（二）：Annotation注解配置方式 ** ###

**1.【注解配置】：**

**AnnotationConfiguration：**

package com.caox.rabbitmq.demo._08_spring_rabbitmq_sync_annotation;
    
    import com.rabbitmq.client.AMQP;
    import org.springframework.amqp.core.AmqpAdmin;
    import org.springframework.amqp.core.Queue;
    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.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * Created by nazi on 2018/7/27.
     */
    @Configuration
    public class AnnotationConfiguration {
    
        //指定队列名称 routingkey的名称默认为Queue的名称，使用Exchange类型为DirectExchange
        protected String springQueueDemo = "create.world.queue";
    
        //创建链接
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory connectionFactory = new CachingConnectionFactory("127.0.0.1");
            connectionFactory.setUsername("caoxia");
            connectionFactory.setPassword("caoxia123456");
            connectionFactory.setPort(AMQP.PROTOCOL.PORT);
            return connectionFactory;
        }
    
        //创建rabbitAdmin 代理类
        @Bean
        public AmqpAdmin amqpAdmin() {
            return new RabbitAdmin(connectionFactory());
        }
    
        //创建rabbitTemplate 消息模板类
        @Bean
        public RabbitTemplate rabbitTemplate() {
            RabbitTemplate template = new RabbitTemplate(connectionFactory());
            //The routing key is set to the name of the queue by the broker for the default exchange.
            template.setRoutingKey(this.springQueueDemo);
            //Where we will synchronously receive messages from
            template.setQueue(this.springQueueDemo);
            return template;
        }
    
        //
        // Every queue is bound to the default direct exchange
        public Queue helloWorldQueue() {
            return new Queue(this.springQueueDemo);
        }
    
    	/*
    	@Bean
    	public Binding binding() {
    		return declare(new Binding(helloWorldQueue(), defaultDirectExchange()));
    	}*/
    
    	/*
    	@Bean
    	public TopicExchange helloExchange() {
    		return declare(new TopicExchange("hello.world.exchange"));
    	}*/
    
    	/*
    	public Queue declareUniqueQueue(String namePrefix) {
    		Queue queue = new Queue(namePrefix + "-" + UUID.randomUUID());
    		rabbitAdminTemplate().declareQueue(queue);
    		return queue;
    	}
    
    	// if the default exchange isn't configured to your liking....
    	@Bean Binding declareP2PBinding(Queue queue, DirectExchange exchange) {
    		return declare(new Binding(queue, exchange, queue.getName()));
    	}
    
    	@Bean Binding declarePubSubBinding(String queuePrefix, FanoutExchange exchange) {
    		return declare(new Binding(declareUniqueQueue(queuePrefix), exchange));
    	}
    
    	@Bean Binding declarePubSubBinding(UniqueQueue uniqueQueue, TopicExchange exchange) {
    		return declare(new Binding(uniqueQueue, exchange));
    	}
    
    	@Bean Binding declarePubSubBinding(String queuePrefix, TopicExchange exchange, String routingKey) {
    		return declare(new Binding(declareUniqueQueue(queuePrefix), exchange, routingKey));
    	}*/
    }

**2【生产者Producer】：**

package com.caox.rabbitmq.demo._08_spring_rabbitmq_sync_annotation;
    
    import org.springframework.amqp.core.AmqpAdmin;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.annotation.AnnotationConfigApplicationContext;
    /**
     * Created by nazi on 2018/7/27.
     */
    
    public class Producer {
        public static void main(String[] args) {
            ApplicationContext context = new AnnotationConfigApplicationContext(AnnotationConfiguration.class);
    
            // 声明创建队列
            AmqpAdmin amqpAdmin = context.getBean(AmqpAdmin.class);
            Queue helloWorldQueue = new Queue("create.world.queue");
            amqpAdmin.declareQueue(helloWorldQueue);
    
            AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
            amqpTemplate.convertAndSend("Hello World");
            System.out.println("Sent: Hello World");
        }
    }

**3.【消费者 Consumer】： **

package com.caox.rabbitmq.demo._08_spring_rabbitmq_sync_annotation;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.annotation.AnnotationConfigApplicationContext;
    
    /**
     * Created by nazi on 2018/7/27.
     */
    
    public class Consumer {
        public static void main(String[] args) {
            ApplicationContext context = new AnnotationConfigApplicationContext(AnnotationConfiguration.class);
            AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
            System.out.println("Received: " + amqpTemplate.receiveAndConvert());
        }
    }

[RabbitMQ_github]: https://github.com/JnCaoxia/rabbitmq