RabbitMQ和Spring的整合

作者: AmeeLove | 来源:发表于2017-11-06 16:25 被阅读70次

spring-RabbitMQ

MAVEN

<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.0</modelVersion>

    <groupId>com.ghghcn</groupId>
    <artifactId>rmq</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>rmq</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spring.version>4.3.12.RELEASE</spring.version>
        <spring-rabbit.version>1.7.4.RELEASE</spring-rabbit.version>
        <slf4j.version>1.7.25</slf4j.version>
        <logback.verison>1.2.3</logback.verison>
        <aspectj.version>1.8.11</aspectj.version>
        <validation.version>1.1.0.Final</validation.version>
        <hibernate-validator.version>5.4.1.Final</hibernate-validator.version>
        <jdk.version>1.8</jdk.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</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-aop</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-aspects</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-tx</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jdbc</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <!--rabbitmq依赖 -->
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.7.4.RELEASE</version>
        </dependency>
        <!--slf4j-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!--乔接器JCL，全称为"Jakarta Commons Logging"，也可称为"Apache Commons Logging"。-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!--log4j to slf4j -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>log4j-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!--java.util.log to slf4j-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jul-to-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!--logback-->
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>${logback.verison}</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-access</artifactId>
            <version>${logback.verison}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

spring.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"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">

    <import resource="classpath:spring/spring-rabbitMQ.xml"/>


    <!-- <aop:aspectj-autoproxy/>-->
  <context:property-placeholder location="classpath:system.properties"/>
    <context:component-scan base-package="com.ghghcn.rmq"/>

    <bean id="springContext" class="com.ghghcn.rmq.context.SpringContextHolder"></bean>

    <!-- 激活annotation功能 -->
    <context:annotation-config />
    <!-- 激活annotation功能 -->
    <context:spring-configured />

</beans>

service 发送者

package com.ghghcn.rmq.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;

@Service
public class MessageProducer {
    private Logger logger = LoggerFactory.getLogger(MessageProducer.class);

    @Resource(name="amqpTemplate")
    private AmqpTemplate amqpTemplate;

    @Resource(name="amqpTemplate2")
    private AmqpTemplate amqpTemplate2;

    public void sendMessage(Object message) throws IOException {
        logger.info("to send message:{}", message);
        amqpTemplate.convertAndSend("queueTestKey", message);
        amqpTemplate.convertAndSend("queueTestChris", message);
        amqpTemplate2.convertAndSend("shijj.xxxx.wsdwd", message);
    }
}

接收者

package com.ghghcn.rmq.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class ChrisConsumer implements MessageListener {
    private Logger logger = LoggerFactory.getLogger(ChrisConsumer.class);


    public void onMessage(Message message) {
        logger.info("chris receive message------->:{}", message);
    }

}

package com.ghghcn.rmq.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class MessageConsumer implements MessageListener {
    private Logger logger = LoggerFactory.getLogger(MessageConsumer.class);


    public void onMessage(Message message) {
        logger.error("consumer receive message------->:{}", message);

    }

}

package com.ghghcn.rmq.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class ShijjConsumer implements MessageListener {
    private Logger logger = LoggerFactory.getLogger(ShijjConsumer.class);

    public void onMessage(Message message) {
        logger.info("shijj receive message------->:{}", message);
    }
}

spring-RabbitMQ.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:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
     http://www.springframework.org/schema/rabbit
     http://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd">
    <!--配置connection-factory，指定连接rabbit server参数 -->
    <rabbit:connection-factory id="connectionFactory" virtual-host="/"
                               username="${rabbit.username}" password="${rabbit.password}" host="${rabbit.address}" port="${rabbit.port}"
    />


    <!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
    <rabbit:admin id="connectAdmin" connection-factory="connectionFactory"/>

    <!--定义queue -->
    <rabbit:queue name="queueTest" durable="true" auto-delete="false"
                  exclusive="false" declared-by="connectAdmin"/>

    <!-- 定义direct exchange，绑定queueTest -->
    <rabbit:direct-exchange name="exchangeTest"
                            durable="true" auto-delete="false" declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!--定义rabbit template用于数据的接收和发送 -->
    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"
                     exchange="exchangeTest"/>

    <!-- 消息接收者 -->
    <bean id="messageReceiver" class="com.ghghcn.rmq.listener.MessageConsumer"></bean>

    <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container
            connection-factory="connectionFactory">
        <rabbit:listener queues="queueTest" ref="messageReceiver"/>
    </rabbit:listener-container>

    <!--定义queue -->
    <rabbit:queue name="queueChris" durable="true"
                  auto-delete="false" exclusive="false" declared-by="connectAdmin"/>

    <!-- 定义direct exchange，绑定queueTest -->
    <rabbit:direct-exchange name="exchangeTest"
                            durable="true" auto-delete="false" declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="queueChris" key="queueTestChris" ></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!-- 消息接收者 -->
    <bean id="receiverChris" class="com.ghghcn.rmq.listener.ChrisConsumer"></bean>

    <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container
            connection-factory="connectionFactory">
        <rabbit:listener queues="queueChris" ref="receiverChris"/>
    </rabbit:listener-container>

    <!-- 分隔线 -->
    <!--配置connection-factory，指定连接rabbit server参数 -->
    <rabbit:connection-factory id="connectionFactory2" virtual-host="user"
                               username="user" password="user" host="127.0.0.1" port="5672"/>

    <!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
    <rabbit:admin id="connectAdmin2" connection-factory="connectionFactory2"/>

    <!--定义queue -->
    <rabbit:queue name="queueShijj" durable="true"
                  auto-delete="false" exclusive="false" declared-by="connectAdmin2"/>

    <!-- 定义direct exchange，绑定queueTest -->
    <rabbit:topic-exchange name="exchangeTest2"
                           durable="true" auto-delete="false" declared-by="connectAdmin2">
        <rabbit:bindings>
            <rabbit:binding queue="queueShijj" pattern="shijj.#"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:topic-exchange>

    <!--定义rabbit template用于数据的接收和发送 -->
    <rabbit:template id="amqpTemplate2" connection-factory="connectionFactory2"
                     exchange="exchangeTest2"/>

    <!-- 消息接收者 -->
    <bean id="recieverShijj" class="com.ghghcn.rmq.listener.ShijjConsumer"></bean>

    <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container
            connection-factory="connectionFactory2">
        <rabbit:listener queues="queueShijj" ref="recieverShijj"/>
    </rabbit:listener-container>
</beans>

测试

package com.ghghcn;

import com.ghghcn.rmq.service.MessageProducer;

import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class TestClass {
    private Logger logger = LoggerFactory.getLogger(TestClass.class);

    private ApplicationContext context = null;

    @Before
    public void setUp() throws Exception {
        context = new ClassPathXmlApplicationContext("spring/spring.xml");
    }


    @Test
    public void should_send_a_amq_message() throws Exception {
        MessageProducer messageProducer = (MessageProducer) context.getBean("messageProducer");

        int a = 100;
        while (a > 0) {
            //amqpTemplate.convertAndSend("Hello, I am amq sender num :" + a--);
            messageProducer.sendMessage("Hello, I am amq sender num :" + a--);

            try {
                //暂停一下，好让消息消费者去取消息打印出来
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }
}

system.properties

rabbit.address=127.0.0.1
rabbit.port=5672
rabbit.username=admin
rabbit.password=admin

RabbitMQ和Spring的整合

spring-RabbitMQ

MAVEN

spring.xml

service 发送者

接收者

spring-RabbitMQ.xml

测试

system.properties

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读