java消息服务之ActiveMQ入门（0x01）

java消息服务之ActiveMQ入门（0x02）

ActiveMQ能做什么

大多数情况下ActiveMQ被用于做系统之间的数据交换。

比如说几个不同的系统之间需要进行业务的联办，这个时候我们就需要一个第三方消息中间件来做消息传递。而ActiveMQ就是这个中间件。

使用ActiveMQ的用途和优点：

1. 将数据从一个应用程序传送到另一个应用程序，或者从软件的一个模块传送到另外一个模块；
2. 负责建立网络通信的通道，进行数据的可靠传送。
3. 保证数据不重发，不丢失
4. 能够实现跨平台操作，能够为不同操作系统上的软件集成数据传送服务

安装使用ActiveMQ

1. 下载ActiveMQ
2. 直接解压，启动，在命令行cd到ActiveMQ/bin目录下 ./activemq start
3. 检测是否已启动

   ActiveMQ默认采用61616端口提供JMS服务，使用8161端口提供管理控制台服务，执行以下命令便可以检验是否已经成功启动ActiveMQ服务

   命令行中：netstat -an|find "61616"

   直接访问ActiveMQ管理页面http://localhost:8161/admin/ 默认用户名密码admin/admin

4. 关闭ActiveMQ可以使用./activemq stop

测试使用ActiveMQ

其实ActiveMQ是按照JMS（java消息服务）规范实现

JMS（java消息服务）就是典型的异步消息处理机制

ActiveMQ默认提供两种通信模式

p2p（点对点）

p2p比较简单，一方发送消息，一方接收消息。相互通信的双方是通过一个类似于队列的方式来进行交流。而在p2p里一个queue只有一个发送者和一个接收者。queue之间是通过名字区别的

//发送消息
public class P2pSend {
    public static void main(String[] args) throws JMSException, InterruptedException {
         //创建链接工厂  
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
        //创建连接  
        Connection connection = connectionFactory.createConnection();
        connection.start();  
         //创建一个会话  
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
        //创建生产者 发送消息的人  
        MessageProducer producer = session.createProducer(null);
        Destination destination = session.createQueue("JOBS.1");  
        Message message = session.createObjectMessage(123);  
        System.out.println("Sending: id: " + ((ObjectMessage)message).getObject() + " on queue: " + destination);  
        producer.send(destination, message);
        
        producer.close();
        session.close();
        connection.close();
    }
}

---

//接收消息
public class P2pReceiver {
    public static void main(String[] args) throws JMSException, InterruptedException {
         //创建链接工厂  
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
        //创建连接  
        Connection connection = connectionFactory.createConnection();
        connection.start();  
         //创建一个会话  
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
        Destination destination = session.createQueue("JOBS.1");  
        MessageConsumer messageConsumer = session.createConsumer(destination); 
        messageConsumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message arg0) {
                 try {  
                        Thread.sleep(2000);
                        System.out.println("1" + " id:" + ((ObjectMessage)arg0).getObject());  
                    } catch (Exception e) {  
                        e.printStackTrace();  
                    }  
            }
        });
        //session.close();
        //connection.close();
    }
}

这里主要就是通过session.createQueue("队列名称")中的队列名称来判断有那个消费者进行消费的。

访问localhost:8161/admin/后台可以看到

publish-subscribe（发布/订阅）

//消息的发送者
public class PubSend {
    public static void main(String[] args) throws JMSException {
         //创建链接工厂  
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
        //创建连接  
        Connection connection = connectionFactory.createConnection();
        connection.start();  
         //创建一个会话  
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
        //创建生产者 发送消息的人  
        MessageProducer producer = session.createProducer(null);
        Destination destination = session.createTopic("JOB01");
        MapMessage message = session.createMapMessage();  
        message.setString("name", "shli");  
        message.setDouble("price", 1.00);  
        message.setBoolean("up", true);  
        producer.send(destination, message); 
        producer.close();
        session.close();
        connection.close();
    }
}

---

//消息接收者
public class PubReceive {
    public static void main(String[] args) throws JMSException {
         //创建链接工厂  
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");  
        //创建连接  
        Connection connection = connectionFactory.createConnection();
        connection.start();  
         //创建一个会话  
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
        Destination destination = session.createTopic("JOB01");
        MessageConsumer messageConsumer = session.createConsumer(destination);
        messageConsumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message arg0) {
                MapMessage map = (MapMessage)arg0;  
                try {
                    String shli = map.getString("name");
                    double price = map.getDouble("price");  
                    boolean up = map.getBoolean("up");  
                    System.out.println(shli + "----" + price +"------"+up);  
                } catch (JMSException e) {
                    e.printStackTrace();
                }  
                
            }
        });
        
        MessageConsumer messageConsumer1 = session.createConsumer(destination);
        messageConsumer1.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message arg0) {
                MapMessage map = (MapMessage)arg0;  
                try {
                    String shli = map.getString("name");
                    double price = map.getDouble("price");  
                    boolean up = map.getBoolean("up");  
                    System.out.println(shli + "-1-1-1-" + price +"-1--1-1--"+up);  
                } catch (JMSException e) {
                    e.printStackTrace();
                }  
                
            }
        });
      //session.close();
      //connection.close();
    }
}

#######这里主要注意的地方就是

• 消费者只能接收到注册之后，发送端发送的消息

• 发送端发送的消息会分发给所有注册的消费者

• 注册发送端的关键在于session.createTopic("Topic名称")中的Topic名称

访问localhost:8161/admin/ 后台可以看到

说明

我也是第一次学习MQ，记录学习内容，难免会有错误的地方。如果发现，请指出，谢谢~

下一篇将介绍 ActiveMQ的本地事务、消费者接收消息时的异步同步以及接收端和发送端相互通信的情况

源码下载