文章目录
- 消费者程序接收消息并通过 WebSocket 将消息传递给前端
消费者程序接收消息并通过 WebSocket 将消息传递给前端
ActiveMQ 是一个开源的消息代理服务，可以用来实现各种消息传递模式，包括点对点和发布/订阅模型。要将数据从 ActiveMQ 推送到前端，通常可以通过以下步骤实现：
- 配置 ActiveMQ 服务器：首先确保 ActiveMQ 服务器已经正确配置和运行。
- 生产者发送消息：生产者程序将数据发送到 ActiveMQ 队列或主题中。
- 消费者接收消息：消费者程序从 ActiveMQ 队列或主题中接收消息，并将其传递给前端。
- 前端接收消息：前端通过 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<>();
@OnOpen
public void onOpen(Session session) {
sessions.add(session);
}
@OnClose
public 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() {
@Override
public 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>
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 并订阅相应的主题，以接收并显示消息。
