使用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.xxx\bin\activemq.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()) + "\n" + 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;
    }
}