消費者程序接收消息并通過 WebSocket 將消息傳遞給前端

ActiveMQ 是一個開源的消息代理服務，可以用來實現各種消息傳遞模式，包括點對點和發布/訂閱模型。要將數據從 ActiveMQ 推送到前端，通常可以通過以下步驟實現：

1. 配置 ActiveMQ 服務器：首先確保 ActiveMQ 服務器已經正確配置和運行。
2. 生產者發送消息：生產者程序將數據發送到 ActiveMQ 隊列或主題中。
3. 消費者接收消息：消費者程序從 ActiveMQ 隊列或主題中接收消息，并將其傳遞給前端。
4. 前端接收消息：前端通過 WebSocket 或者其他方式接收從消費者程序傳遞過來的消息。
   下面是一個具體的實現方案：
   步驟 1：配置 ActiveMQ 服務器
   確保 ActiveMQ 服務器已經正確配置和運行。可以下載 ActiveMQ 的二進制文件并啟動服務。
   ./bin/activemq start
   步驟 2：生產者發送消息
   使用 Java 作為示例，生產者代碼如下：

import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class Producer {public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");Connection connection = factory.createConnection();connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination destination = session.createQueue("TEST.QUEUE");MessageProducer producer = session.createProducer(destination);TextMessage message = session.createTextMessage("Hello, World!");producer.send(message);session.close();connection.close();}
}

步驟 3：消費者接收消息
消費者程序接收消息并通過 WebSocket 將消息傳遞給前端。

import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;@ServerEndpoint("/websocket")
public class Consumer {private static CopyOnWriteArraySet<Session> sessions = new CopyOnWriteArraySet<>();@OnOpenpublic void onOpen(Session session) {sessions.add(session);}@OnClosepublic void onClose(Session session) {sessions.remove(session);}public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");Connection connection = factory.createConnection();connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination destination = session.createQueue("TEST.QUEUE");MessageConsumer consumer = session.createConsumer(destination);consumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {try {String text = ((TextMessage) message).getText();for (Session s : sessions) {s.getBasicRemote().sendText(text);}} catch (JMSException | IOException e) {e.printStackTrace();}}}});}
}

步驟 4：前端接收消息
前端代碼（例如，HTML + JavaScript）：

Messages

<script>var socket = new WebSocket("ws://localhost:8080/websocket");socket.onmessage = function(event) {var messages = document.getElementById('messages');var message = document.createElement('li');message.textContent = event.data;messages.appendChild(message);};
</script>

通過以上步驟，數據可以從 ActiveMQ 推送到前端。生產者將消息發送到 ActiveMQ 隊列，消費者從隊列中接收消息并通過 WebSocket 將消息傳遞給前端，前端通過 WebSocket 接收并顯示消息。 ## 使用 MQTT 協議將數據推送到前端 ActiveMQ 也支持 MQTT 協議，可以使用 MQTT 協議將數據推送到前端。以下是一個使用 ActiveMQ 和 MQTT 將數據推送到前端的示例。 步驟 1：配置 ActiveMQ 以支持 MQTT 首先，確保 ActiveMQ 服務器已經正確配置和運行。ActiveMQ 默認支持 MQTT，但需要確保相關的 MQTT 連接器已啟用。在 conf/activemq.xml 文件中，確認以下配置已存在： 啟動 ActiveMQ 服務器： ./bin/activemq start 步驟 2：生產者發送消息（使用 MQTT 協議） 生產者代碼可以使用任意支持 MQTT 的庫，例如 Eclipse Paho MQTT 客戶端庫。以下是一個使用 Java 的示例：

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;public class MqttProducer {public static void main(String[] args) {String broker = "tcp://localhost:1883";String topic = "test/topic";String content = "Hello, MQTT!";int qos = 2;String clientId = "JavaProducer";try {MqttClient client = new MqttClient(broker, clientId);client.connect();MqttMessage message = new MqttMessage(content.getBytes());message.setQos(qos);client.publish(topic, message);client.disconnect();} catch (MqttException e) {e.printStackTrace();}}
}

步驟 3：前端接收消息（使用 MQTT 協議）
前端可以使用 MQTT.js 庫，通過 WebSocket 連接到 ActiveMQ 服務器并接收消息。需要確保 ActiveMQ 服務器支持 MQTT over WebSocket。可以在 conf/jetty.xml 文件中添加以下配置：

<bean id="mqttWS" class="org.eclipse.jetty.server.nio.SelectChannelConnector"><property name="host" value="0.0.0.0"/><property name="port" value="1884"/><property name="acceptors" value="2"/><property name="maxIdleTime" value="30000"/><property name="lowResourcesMaxIdleTime" value="1500"/><property name="lowResourcesConnections" value="50"/><property name="statsOn" value="false"/><property name="confidential" value="false"/><property name="lowResourcesMaxIdleTime" value="30000"/><property name="acceptQueueSize" value="100"/><property name="maxBuffers" value="2048"/><property name="requestHeaderSize" value="8192"/><property name="responseHeaderSize" value="8192"/>
</bean>
在前端 HTML 中使用 MQTT.js 庫：
<!DOCTYPE html>
<html>
<head><title>ActiveMQ MQTT WebSocket Demo</title><script src="https://cdnjs.cloudflare.com/ajax/libs/paho-mqtt/1.0.1/mqttws31.min.js"></script>
</head>
<body><h1>Messages</h1><ul id="messages"></ul><script>var client = new Paho.MQTT.Client("localhost", 1884, "webClient");client.onMessageArrived = function(message) {var messages = document.getElementById('messages');var msg = document.createElement('li');msg.textContent = message.payloadString;messages.appendChild(msg);};client.connect({onSuccess: function() {client.subscribe("test/topic");}});</script>
</body>
</html>

通過以上步驟，數據可以通過 MQTT 協議從 ActiveMQ 推送到前端。生產者將消息發布到 MQTT 主題，前端通過 MQTT over WebSocket 連接到 ActiveMQ 并訂閱相應的主題，以接收并顯示消息。