ActiveMQ-与Spring整合-队列篇

Posted 2023-02-19 闲言博客

1.导入依赖

    <dependencies>
        <!-- activemq所需要的jar包 -->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.15.9</version>
        </dependency>
        <dependency>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-spring</artifactId>
            <version>4.17</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.11.3</version>
        </dependency>
        <!-- activeMQ对JMS的支持，整合Spring和ActiveMQ -->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>5.2.10.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.9</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

2.创建spring配置文件

spring-activemq.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.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd">
    
    <!-- 扫描包 -->
    <context:component-scan base-package="cn.bloghut" />

    <!-- 配置生产者 -->
    <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
        <property name="connectionFactory" ref="mqUrl" />
        <!--最大连接数-->
        <property name="maxConnections" value="100" />
    </bean>

    <!-- 真正可以生产Connection的ConnectionFactory,对应的JMS服务厂商提供 -->
    <bean id="mqUrl" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://127.0.0.1:61616" />
    </bean>

    <!-- 这个是队列目的地，点对点的 -->
    <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg index="0" value="spring-active-queue" />
    </bean>

    <!-- Spring提供的JMS工具类，他可以进行消息发送，接收等 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!--绑定工厂-->
        <property name="connectionFactory" ref="jmsFactory" />
        <!--绑定队列-->
        <property name="defaultDestination" ref="destinationQueue" />
        <property name="messageConverter" >
            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter" />
        </property>
    </bean>
</beans>

3.创建消息生产者

public interface SpringMQ_Produce 
	//生产消息
	void send(String messageStr);


@Service("springProduce")
public class SpringProduceImpl implements SpringProduce 

	@Autowired
	private JmsTemplate jmsTemplate;
	
	public void send(String messageStr) 
		jmsTemplate.send((session)->
			TextMessage textMessage = session.createTextMessage(messageStr);
			return textMessage;
		);
	

4.创建消息消费者

public interface SpringConsumer 
	//消费消息
	void read();


@Service("springConsumer")
public class SpringConsumerImpl implements SpringConsumer

    @Autowired
    private JmsTemplate jmsTemplate;

    @Override
    public void read() 
        while (true)
            Object o = jmsTemplate.receiveAndConvert();
            if (null != o)
                String data = (String) o;
                System.out.println(data);
            
        
    

5.创建测试发送消息方法

public class App 
    ApplicationContext atx = null;
    private SpringProduce sp = null;
    private SpringConsumer sc = null;
    
        atx = new ClassPathXmlApplicationContext("spring-activemq.xml");
        sc = (SpringConsumer) atx.getBean("springConsumer");
        sp = (SpringProduce) atx.getBean("springProduce");
    

    @Test
    public void testProduct() 
        for (int i = 1;i <=3; i++)
            sp.send("队列消息发送成功!!! >"+i);
        
    
    @Test
    public void  testConsumer()
        sc.read();
    

6.启动生产者

testProduct()

已经发现有3条待消费

7.启动消费者

testConsumer()

有1个消费者 消费了3条消息。