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

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

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

解压完成后找到指定路径

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

 访问本地管理页面http://127.0.0.1:8161/

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

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

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

 先看生产者

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

1.创建一个java项目

2.创建lib目录

3.引入mqttclientjar包

目录结构可如下：

 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.执行代码结果

 6.查看平台中数据

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();
	}

接收数据 

当前链接