activemq 使用介绍

1，activemq 分为 queue和topic 两种

2，下面先介绍queue ，使用spring 集成

build.gradle

1compile "org.springframework:spring-jms:3.2.1.RELEASE"
2    compile "org.apache.activemq:activemq-core:5.7.0"
3    compile "ch.qos.logback:logback-core:1.0.9"
4    compile "ch.qos.logback:logback-classic:1.0.9"
5    compile "ch.qos.logback:logback-access:1.0.9"
6    testCompile group: 'junit', name: 'junit', version: '4.10'

applicationContext.xml

1<description>Spring公共配置</description>
2
3    <!-- 使用annotation 自动注册bean,并检查@Required,@Autowired的属性已被注入 -->
4    <context:component-scan base-package="net.tt64"/>
5    <!-- ActiveMQ 连接工厂 -->
6    <bean id="advancedConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
7        <property name="brokerURL" value="vm://cheyoushuo" />
8        <property name="useAsyncSend" value="true" />
9    </bean>
10    
11    <!-- Spring Caching 连接工厂 -->
12    <bean id="advancedCachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
13        <property name="targetConnectionFactory" ref="advancedConnectionFactory" />
14        <property name="sessionCacheSize" value="10" />
15    </bean>
16    
17    <!-- Queue定义 -->
18    <bean id="notifyQueue" class="org.apache.activemq.command.ActiveMQQueue">
19        <constructor-arg value="test.jms" />
20    </bean>
21
22    <!-- Spring JMS Template -->
23    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
24        <property name="connectionFactory" ref="advancedCachingConnectionFactory" />
25        <!-- 使 deliveryMode, priority, timeToLive设置生效-->
26        <property name="explicitQosEnabled" value="true" />
27        <!-- 设置NON_PERSISTENT模式, 默认为PERSISTENT -->
28        <property name="deliveryPersistent" value="true" />
29        <!-- 设置优先级, 默认为4 -->
30        <property name="priority" value="9" />
31    </bean>
32    
33    <!-- 异步接收Queue消息Container -->
34    <bean id="advancedQueueContainer" depends-on="jmsService" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
35        <property name="connectionFactory" ref="advancedConnectionFactory" />
36        <property name="destination" ref="notifyQueue" />
37        <property name="messageListener" ref="jmsService" />
38        <!-- 初始5个Consumer, 可动态扩展到10 -->
39        <property name="concurrentConsumers" value="1" />
40        <property name="maxConcurrentConsumers" value="1" />
41        <!-- 设置消息确认模式为Client -->
42        <property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE" />
43    </bean>

jms sender 

1@Autowired
2    private JmsTemplate jmsTemplate;
3    @Autowired
4    private Destination dest;
5
6    public void send(final Serializable obj) {
7        jmsTemplate.send(dest, new MessageCreator() {
8            @Override
9            public Message createMessage(Session session) throws JMSException {
10                ObjectMessage msg = session.createObjectMessage(obj);
11                return msg;
12            }
13        });
14    }

jms receiver

1@Override
2    public void onMessage(Message message) {
3        if (message instanceof ObjectMessage) {
4            ObjectMessage om = (ObjectMessage) message;
5            try {
6                System.out.println(om.getObject().toString() + " recieved");
7                Thread.currentThread().sleep(1000);
8                System.out.println(om.getObject().toString() + "  process over");
9            } catch (InterruptedException ex) {
10            } catch (JMSException ex) {
11            }
12        }
13    }

Main

1public static void main(String[] args) {
2        ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
3        JmsSender sender = ctx.getBean(JmsSender.class);
4        for (int i = 0; i < 100; i++) {
5            System.out.println("message " + i + "send");
6            sender.send("message" + i);
7        }
8    }

这里将 concurrentConsumers 调为 1, 这就是单任务队列