ActiveMq简单介绍
一般常见的消息中间件有:RabbitMQ,ActiveMq,RocketMQ(阿里)等，都称之为MQ(Message Queue，消息队列)，这里介绍ActiveMQ。
ActiveMQ是Apache出品，最流行的，能力强劲的开源消息总线。从设计上保证了高性能的集群，客户端-服务器，点对点。完全支持JMS1.1和J2EE1.4规范的JMSProvider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。
特性：
⒈多种语言和协议编写客户端。语言:Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议：OpenWire,StompREST,WSNotification,XMPP,AMQP
⒉完全支持JMS1.1和J2EE1.4规范（持久化，XA消息，事务)
⒊对Spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去，而且也支持Spring2.0的特性
⒋通过了常见J2EE服务器（如Geronimo,JBoss4,GlassFish,WebLogic)的测试，其中通过JCA1.5resourceadaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE1.4商业服务器上
⒌支持多种传送协议：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
⒍支持通过JDBC和journal提供高速的消息持久化
⒎从设计上保证了高性能的集群，客户端-服务器，点对点
⒏支持Ajax
⒐支持与Axis的整合
⒑可以很容易得调用内嵌JMSprovider，进行测试
消息形式：
一种是点对点的，即一个生产者和一个消费者一一对应；
另一种是发布/订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。
JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。
• StreamMessage – Java原始值的数据流
• MapMessage–一套名称-值对
• TextMessage–一个字符串对象
• ObjectMessage–一个序列化的 Java对象
• BytesMessage–一个字节的数据流
ActiveMq的安装启动
ActiveMQ的安装启动很简单。
下载：进入http://activemq.apache.org/下载ActiveMQ
这里在linux下进行安装。
1、将下载的ActiveMQ压缩包上传到linux（我是上传到了/usr目录下），用远程连接工具（比如ssh）。
2、解压缩 ：tar zxvf apache-activemq-5.15.4-bin.tar.gz
3、启动：进入bin目录 ./activemq start
查看是否启动成功，./activemq status
如果没有启动成功，那么就要修改hosts文件了，配置机器名和本机（127.0.0.1）的映射关系。比如本机名是laowang（可以使用命令/etc/sysconfig/network查看机器名称）。那么需要做如下的修改：
[root@laowang bin]# cat /etc/hosts
127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
::1 localhost localhost.localdomain localhost6 localhost6.localdomain6
127.0.0.1 laowang.localdomain itheima32//加上这一行
然后浏览器访问：192.168.25.128:8161(8161是固定端口号)
image.png
那么说明启动成功了。
注：点击Manage ActiveMqQ broker输入用户名密码为admin。
ActiveMQ在java中的使用
image.png
对照着上面两种模式图，编写代码的时候后会更容易理解。
创建个maven项目，导入的包是
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.14.0</version>
</dependency>
包版本超过5.14.0会和mybatis冲突
1、点对点模式：生产者
/*
* 测试点对点模式：生产者
*/
@Test
public void testQueueProducer() throws JMSException{
//1.创建一个连接工厂对象,需要指定服务的ip及端口号
ActiveMQConnectionFactory factory =
new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
//2.使用工厂对象来创建一个Connection对象
Connection connection = factory.createConnection();
//3.开启连接，调用Connection对象的start方法
connection.start();
//4.创建一个Session对象
//第一个参数:是否开启事务,如果开启事务true带二个参数无意义，一般不开启事务false
//第二个参数:应答模式。一般自动应答或者手动应答。一般自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.使用Session对象创建一个Destination对象。两种形式queue,topic,现在应该使用queue
Queue queue = session.createQueue("测试消息队列First");
//6.使用Session对象创建一个Producer对象
MessageProducer producer = session.createProducer(queue);
//7.创建一个Message对象，可以使用TextMessage
Message message = session.createTextMessage("hello activemq!");
//8.发送消息
producer.send(message);
//9.关闭资源
producer.close();
session.close();
connection.close();
}
运行方法后，会发送一个消息，可以在页面查看。
image.png
image.png
一条挂起的消息，一条入队的消息,也能看到目的地跟消息详情，以及消息被持久化。
2、点对点模式：消费者
前面生产了一条消息，这里就测试消费者去消费这条消息。
/*
* 测试点对点模式：消费者
*/
@Test
public void testQueueConsumer() throws Exception{
//1.创建连接工厂
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
//2.创建连接
Connection connection = factory.createConnection();
//3.开启连接
connection.start();
//4.创建session对象
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.创建目的地，需要跟生产者发送消息的目的地一致不然不能接收到消息
Queue queue = session.createQueue("测试消息队列First");
//使用Session对象创建一个消费者对象
MessageConsumer consumer = session.createConsumer(queue);
//接收消息
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage)message;
try {
String text = textMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//等待接收消息,消费完了就没了
System.in.read();//等待我们敲击键盘才会向下执行
//关闭资源
consumer.close();
session.close();
connection.close();
}
运行后
image.png
image.png
被挂起的消息已经被消费了，一个消费者，一个入队消息一个出队消息。如果再发生消息的话控制台会持续（也就是消费者能持续接收生产者发送的消息）输出直至按回下回车键才结束。这个时候消费者数量也变为0了。
3、发布/订阅模式：生产者
跟点对点模式的生产者很类似，主要是创建的目的地为Topic而不是Queue了。
/*
* 发布订阅模式：生产者
*/
@Test
public void testTopicProducer() throws Exception{
//1.创建连接工厂
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
//2.创建连接
Connection connection = factory.createConnection();
//3.开启连接
connection.start();
//4.创建session对象
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.创建目的地Destination，该模式创建的是Topic
Topic topic = session.createTopic("测试消息队列First");
//6.创建生产者
MessageProducer producer = session.createProducer(topic);
//7.创建消息
TextMessage message = session.createTextMessage("hello topic");
//8.发送消息
producer.send(message);
//9.关闭资源
producer.close();
session.close();
connection.close();
}
运行后去页面点击Topics查看
image.png
0个消费者，该模式默认消息不被持久化，但是可以修改为持久化，只有一条入队消息。
4、发布/订阅模式：消费者
/*
* 发布订阅模式：消费者
*/
@SuppressWarnings("unused")
@Test
public void testTopicConsumer() throws Exception{
//1.创建连接工厂
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.25.128:61616");
//2.创建连接
Connection connection = factory.createConnection();
//3.开启连接
connection.start();
//4.创建session对象
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.创建目的地Destination
Topic topic = session.createTopic("测试消息队列First");
//6.创建消费者
MessageConsumer consumer = session.createConsumer(topic);
//7.接收消息
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage)message;
//获取信息并打印
try {
String text = textMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
System.out.println("----消费者1准备就绪----");
System.in.read();
//9.关闭资源
consumer.close();
session.close();
connection.close();
}
虽然前面发布了一条消息但是既然不能被持久化，那么这里自然就不能去接收该消息了。
所以需要先执行该方法，使消费者处于持续能接收消息的状态，这样生产者每次发送消息都能被接收到。
这里该测试方法运行三次。打印依次为：
System.out.println(“—-消费者1准备就绪—-“);
System.out.println(“—-消费者2准备就绪—-“);
System.out.println(“—-消费者3准备就绪—-“);
就等于有三个消费者了。这个时候再执行测试生产者的方法（执行三次，等于发了三次消息），就能发现三个消费者都接收到了三条消息。
查看控制台：
----消费者1准备就绪----
hello topic
hello topic
hello topic
----消费者2准备就绪----
hello topic
hello topic
hello topic
----消费者3准备就绪----
hello topic
hello topic
hello topic
再看页面：
image.png
4条入队消息（其中一条没被消费），9条出队消息（3个消费者每个消费3条消息）。
网友评论