简介
本文达到的目的是会用。
安装服务器:
官网下载win版的,也可是linux版的;
启动服务器:win版的执行activemq.bat
看看是否安装成功:http://localhost:8161/
activeMQ是一个消息服务器,
2中模式:点对点、广播
点对点:消息进入队列后,只会被接收一次,接收完了就在消息队列里消失了。
广播:采用的是广播订阅的方式,所有的订阅了此主题的,当此主题有消息时,都会接收到
可靠性:
mq会将信息存储
/*
* 消息的类型(消息体):
* TextMessage
* ObjectMessage: Java中的可序列化对象
* StreamMessage: Java中的输入输出流
* BytesMessage: 字节流
* MapMessage: 名/值对的集合,名是String对象,值类型可以是Java任何基本类型
*/
/*
* 术语:
* ConnectionFactory:连接工厂,JMS 用它创建连接;
* Destination:消息的目的地;
* Session:会话,一个发送或接收消息的线程;
*/
发送消息到消息队列:
场景:程序中记录日志,异步的把日志输出等。
配置文件:
Spring-context.xml:
配置jmsConnectionFactory
<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
//value的值就是mq服务器的地址
<property name="brokerURL" value=" tcp://10.128.11.128:61616 " />
<property name="useAsyncSend" value="true" />//表示异步传送,默认是false,同步
</bean>
配置pooledJmsConnectionFactory:池
<bean id="pooledJmsConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
destroy-method="stop">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="maxConnections" value="100" />
<property name="maximumActive" value="50" />
</bean>
配置jmsTemplate:模板,这是核心
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="pooledJmsConnectionFactory" />
</bean>
定义发送消息的队列:
<bean id="sendGSMMessageQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="gsm.message.push.queue" />//队列的名字是任意写的,只要不重复
</bean>
定义生产者:将消息发送到这个队列
<bean id="notificationSender" class="com.gome.gsm.jms.NotificationSender">
<property name="jmsTemplate" ref="jmsTemplate"></property>
//将消息发送到这个队列
<property name="queue" ref="sendGSMMessageQueue"></property>
</bean>
NotificationSender.java:
就是普通的类,注入jmsTemplate和queue,并且有个send方法,并调用jmsTemplate的send方法
//这里发送的是String型的数据
public void send(final String message){
this.jmsTemplate.send(queue, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
logger.info("-------发送消息到消息服务器通知队列成功--" + message + "-------");
}
以上,便实现了将消息发送到消息队列。
既然有了生产者,那么也应该有消费者,否则消息将一直阻塞。
从消息队列接收消息
注:这里不是在那个队列里接收的消息
------------------------------------分割线-----------------------------------------
进行消息的接收
配置receiveMessageQueue:接收消息队列(是否需要定义接收消息队列与项目需求有关,比如我就是接收我前面已经定义的消息发送队列的消息,那么这里也就没有必要定义这个队列了)
<bean id="receiveMessageQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="gsm.message.receiver.queue" />//要在这个消息队列接收消息
</bean>
配置消息监听器:
<bean id="notificationListener" class="com.gome.gsm.jms.NotificationListener">
//这里需要根据系统具体写,因为监听器类里面注入了这个资源
<property name="notificationService" ref="notificationService" />
</bean>
配置消息接收客户端:
<bean id="notificationConsumer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="pooledJmsConnectionFactory" />
<property name="destination" ref="receiveMessageQueue" />
<property name="messageListener" ref="notificationListener" />
</bean>
其中bean 的id为notificationService的类:(项目时注解开发)
编写监听器类:(在消息队列里接收消息的)
public class NotificationListener implements MessageListener {
private static final Logger logger = LoggerFactory.getLogger(NotificationListener.class);
private NotificationService notificationService;
public NotificationService getNotificationService() {
return notificationService;
}
public void setNotificationService(NotificationService notificationService) {
this.notificationService = notificationService;
}
@Override
public void onMessage(Message message) {
try {
if(message instanceof TextMessage){
TextMessage receiveMessage = (TextMessage) message;
String jsonStr = receiveMessage.getText();
logger.info("----接收到通知消息: " + jsonStr + "--------------------");
Map map = JsonUtil.jsonStringToMap(jsonStr);
//根据接收到的消息具体做些事情即可。。。
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
注意:
因为项目中只用到了这种点对点的方式队列,所以这里只有这个。
发送的消息队列可以使可接受的消息队列是一个。具体看需求。
mq就是一个消息服务器,有生产者将消息放到队列里面,至于是谁接收消息,谁是消费者,并不清楚;
参考:http://yinbinhome.iteye.com/blog/1273228