activemq安装windows环境，数据订阅与发布，生产者与消费者

朱雀 2024-03-30 11:05 48阅读 0赞

官网下载，选择适合本地jdk编译环境的部署包

![66878bc840df4b4b8410b07f502eeb68.png][]

安装完成后解压到指定目录，目录不要包含中文，否则可能启动不成功。

解压完成后找到指定路径

![aaaf35e471624dbdb8315705f8138473.png][]

双击即可运行 ，见到如下界面启动成功

![342312b1eeed4d488641ff1490f4e307.png][]

访问本地管理页面[http://127.0.0.1:8161/][http_127.0.0.1_8161]

输入账号密码进入管理页面

![eeb0c01d52c84c04bdb0a9c4185c30a9.png][]

这时候还没有客户端连接，需要编写一个生产者，生产数据

再编写一个客户端，消费数据

> 先看生产者

因为activemq支持多种消息协议，这里我们选择mqtt

## 1.创建一个java项目 ##

### 2.创建lib目录 ###

## 3.引入mqttclientjar包 ##

目录结构可如下：

![1d2a243221bb45d1b12ff5eb2a513ae0.png][]

## 4.编写生产者代码 ##

private static void pub(MqttClient sampleClient, String msg,String topic) 
    			throws MqttPersistenceException, MqttException {
    		MqttMessage message = new MqttMessage(msg.getBytes());
    		message.setQos(qos);
    		message.setRetained(false);
    		  sampleClient.publish(topic, message);
    		  sampleClient.setCallback(new PubCallback());
    //		  MqttTopic top = sampleClient.getTopic(topic);
    //		  top.publish(message);
    		
    	}
    	
    	public static void publish(String str,String clientId,String topic) throws MqttException{
    		MqttClient mqttClient = new SubMsg().connect(clientId);
     
    		if (mqttClient != null) {
    			pub(mqttClient, str, topic);
    			System.out.println("生产者发布："+topic+"主题消息=" + str);
    		}
     
    //		if (mqttClient != null) {
    //			mqttClient.disconnect();
    //		}
    	}

## 5.执行代码结果 ##

![2dd3a8a1fac845bba70611eb1614cf4e.png][]

## 6.查看平台中数据 ##

![afd68c3bb20f4598acfb3e591829a139.png][]

## 7. 编写数据消费端代码 ##

链接并订阅主题

SubMsg.runsub(PropertiesUtil.activemq_clientname,PropertiesUtil.activemq_topic);

接收生产者发布的数据

public void messageArrived(String topic, MqttMessage message) throws Exception {
    		String msg = new String(message.getPayload(),"utf-8");
    		System.out.println("客户端"+PropertiesUtil.activemq_clientname+"（订阅者）接收到"+topic+"主题消息:" + msg);
    		SimpleDateFormat format=new SimpleDateFormat("yyMMdd");
    		File file=new File(PropertiesUtil.file_dir+format.format(System.currentTimeMillis())+".txt");
    		File dir = new File(PropertiesUtil.file_dir);
    		if(!dir.exists())dir.mkdirs();
    		FileOutputStream fos = new FileOutputStream(file, true);
    		fos.write(("客户端"+PropertiesUtil.activemq_clientname+"（订阅者）接收到"+topic+"主题消息：\n"+msg+"\n").getBytes("UTF-8"));
    		fos.close();
    	}

接收数据

![a2328d46e07b475d93fddf8ac533f038.png][]

当前链接

![453ca390a8b2467bb91168eb1e7d61fa.png][]

[66878bc840df4b4b8410b07f502eeb68.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/29bb7f9c0aba4d3493fa19768befff5d.png
[aaaf35e471624dbdb8315705f8138473.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/eac0b6b52e8f440eb716ccbbf88be4dd.png
[342312b1eeed4d488641ff1490f4e307.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/660768aa8e584c1695a3eae78150fb4d.png
[http_127.0.0.1_8161]: http://127.0.0.1:8161/
[eeb0c01d52c84c04bdb0a9c4185c30a9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/e1ae50c1d5284a0286a5ec7b8a7bca83.png
[1d2a243221bb45d1b12ff5eb2a513ae0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/9e93bfab871d4f16b07dac8178245850.png
[2dd3a8a1fac845bba70611eb1614cf4e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/e88216be50764cbcb642e78cd696d501.png
[afd68c3bb20f4598acfb3e591829a139.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/21892a34f9c54108936eeb0476cd51b2.png
[a2328d46e07b475d93fddf8ac533f038.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/75222b91b646457fada19dfcb0fdab2e.png
[453ca390a8b2467bb91168eb1e7d61fa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/e74b70576d2747eda5a671d7eb4ea04a.png