第一课 ActiveMQ的安装及使用

2018-08-27  本文已影响6人  Arroganter

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

image.png
那么说明启动成功了。
注:点击Manage ActiveMqQ broker输入用户名密码为admin。
ActiveMQ在java中的使用


image.png
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

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条消息)。

Apache ActiveMQ 各个版本所依赖的JDK版本

https://blog.csdn.net/yanfeipei/article/details/75267384

上一篇下一篇

猜你喜欢

热点阅读