使用ActiveMQ实现简易聊天功能

一 什么是消息队列

我们可以把消息队列比作是一个存放消息的容器，当我们需要使用消息的时候可以取出消息供自己使用。消息队列是分布式系统中重要的组件，使用消息队列主要是为了通过异步处理提高系统性能和削峰、降低系统耦合性。目前使用较多的消息队列有ActiveMQ，RabbitMQ，Kafka，RocketMQ

二 为什么要用消息队列

使用消息队列主要有两点好处：

1.通过异步处理提高系统性能（削峰、减少响应所需时间）;

2.降低系统耦合性。如果在面试的时候你被面试官问到这个问题的话，一般情况是你在你的简历上涉及到消息队列这方面的内容，这个时候推荐你结合你自己的项目来回答。

三 ActiveMQ

ActiveMQ 是基于 JMS 规范实现的。

JMS消息列队有两种消息模式，一种是点对点的消息模式，还有一种是订阅的模式。

四 实现

ActiveMQ下载地址：http://activemq.apache.org/components/classic/download/

解压缩apache-activemq-5.xxx-bin.zip到一个目录

启动ActiveMQ：运行C: apache-activemq-5.xxxinactivemq.bat

浏览器中输入：http://localhost:8161/admin/ 测试启动情况

使用点对点方式实现聊天功能

编写消息发送类和接收类。发送类中需要连接ActiveMQ 服务器，创建队列，发送消息；接收类中需要ActiveMQ 服务器，读取发送者发送消息所用的队列。接收类实现为一个单独的线程，使用监听器模式，每隔一段时间侦听是否有消息到来，若有消息到来，将消息添加到辅助类消息列表中。

使用2个队列，即对于每一个用户来说，发送消息为一个队列，接受消息为一个队列。

效果如下：

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

import static java.lang.Thread.sleep;

public class MessageReceiver implements Runnable{
    private String url;
    private String user;
    private String password;
    private final String QUEUE;
    private Boolean stop;
    Connection connection;

    public MessageReceiver(String queue, String url, String user, String password) {
        this.url = url;
        this.user = user;
        this.password = password;
        this.QUEUE = queue;
        stop = false;
    }

    public void run() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
        try {
            connection = connectionFactory.createConnection();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination receiveQueue = session.createQueue(QUEUE);
            MessageConsumer consumer = session.createConsumer(receiveQueue);
            connection.start();
            while(!stop) {
                consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        try {
                            //获取到接收的数据
                            String text = ((TextMessage) message).getText();
                            MessageText.setMsg(text);
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                });
                sleep(500);
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }catch (InterruptedException e) {
            //Thread.currentThread().interrupt();
            e.printStackTrace();
        }
    }

    public void setStop(Boolean stop) {
        this.stop = stop;
    }

    public void closeConnection(){
        try {
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }
}

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

public class MessageSender {
    private String url;
    private String user;
    private String password;
    private final String QUEUE;
    private Connection connection;
    private Session session;
    private Destination sendQueue;
    private MessageProducer sender;
    private TextMessage outMessage;
    private DateFormat df;

    public MessageSender(String queue, String url, String user, String password) {
        this.url = url;
        this.user = user;
        this.password = password;
        this.QUEUE = queue;
    }

    public void init() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
        df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            connection = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            sendQueue = session.createQueue(QUEUE);
            sender = session.createProducer(sendQueue);
            outMessage = session.createTextMessage();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void sendMessage(String messageStr) {
        try {
            outMessage = session.createTextMessage();
            String sendStr = df.format(new Date()) + "
" + QUEUE + ": " + messageStr;
            outMessage.setText(sendStr);
            sender.send(outMessage);
            session.commit();
            MessageText.setMsg(sendStr);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void closeConnection() {
        try {
            sender.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }
}