springboot集成rabbitmq和netty

springboot集成rabbitmq和netty"/>

springboot集成rabbitmq和netty

集成rabbitmq（erlang语言）

一、安装软件

1.下载安装如下软件

2.配置环境变量 erl安装目录/bin   rabbtit安装目录/sbin

3.安装插件

运行命令rabbitmq-plugins enable rabbitmq_management 开启Web管理插件

二、登录后台

通过浏览器访问http://localhost:15672，并通过默认用户guest进行登录，密码也是guest，登录后的页面：

2.添加队列

添加完之后如下

3.添加账户，说明：guest账号在本地可以用，到阿里云等外网就不能用guest。新建账号

需要分配权限：

三.配置java

服务器：

1.添加依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.properties文件：

spring.application.name=spirng-boot-rabbitmq-sender
spring.rabbitmq.host=外网ip
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin

3.配置队列：

SenderConf.java

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class SenderConf {@Beanpublic Queue queue() {return new Queue("queue");}
}

4.定义发送接口：

HelloSender.java

import com.zwh.entity.ZwhPolice;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;@Component
public class HelloSender {@Autowiredprivate AmqpTemplate template;public void send(ZwhPolice police) {template.convertAndSend("queue",police);}
}

5.发送信息

@Autowired
private HelloSender helloSender;
//超温报警
ZwhPolice police1 = new ZwhPolice();
police1.setStationId(stationId);
police1.setCreateTime(fmt.format(new Date()));
police1.setPoliceValue(temperetureData);
police1.setPoliceSetValue(valuePoliceValue.toString());
police1.setDeviceName(deviceName + str);
police1.setPoliceLevel("严重");
police1.setPoliceType("高温报警");
police1.setPoliceDescribe(deviceName + str + "温度" + temperetureData + "度,超过设置的温度阈值" + valuePoliceValue + "度");
//发送helloSender.send(police1);

客户端：

重复配置服务器1，2,3步骤

4.配置监听类:

HelloReceive.java

import com.zwh.controller.webSocket.SocketEventHandler;
import com.zwh.entity.TbAdmin;
import com.zwh.entity.ZwhPolice;
import com.zwh.mapper.TbAdminMapper;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.HashMap;
import java.util.List;
import java.util.Map;@Component
public class HelloReceive {@RabbitListener(queues="queue")    //监听器监听指定的Queuepublic void processC(ZwhPolice police) {System.out.println(police.toString());}
}

阿里云需要开放15672和5672端口

 

集成netty-sock.io

1.添加依赖：

 <!--webSocket--><dependency><groupId>com.corundumstudio.socketio</groupId><artifactId>netty-socketio</artifactId><version>1.7.7</version><scope>compile</scope></dependency>

2.配置配置类：

SocketConfig.java

import com.corundumstudio.socketio.Configuration;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.annotation.SpringAnnotationScanner;
import org.springframework.context.annotation.Bean;@org.springframework.context.annotation.Configuration
public class SocketConfig {@Beanpublic SocketIOServer socketIOServer() {Configuration config = new Configuration();//config.setHostname("localhost");config.setPort(8082);return new SocketIOServer(config);}@Beanpublic SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {return new SpringAnnotationScanner(socketServer);}
}

//config.setHostname("localhost");特此说明，部署到阿里云该句要注释掉，负责会报错。本地开发要解除注释

3.配置事件类

SocketEventHandler.java

import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.annotation.OnConnect;
import com.corundumstudio.socketio.annotation.OnDisconnect;
import com.corundumstudio.socketio.annotation.OnEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.HashMap;
import java.util.Map;@Component
public class SocketEventHandler {private Map<String, Object> socketMap = new HashMap<>();@Autowiredprivate SocketIOServer server;@OnConnectpublic void onConnect(SocketIOClient client) {String userId = client.getHandshakeData().getSingleUrlParam("userId");socketMap.put(userId, client);}@OnDisconnectpublic void onDisConnect(SocketIOClient client) {String[] userId = new String[1];socketMap.forEach((key, value) -> {if (value == client) userId[0] = key;});socketMap.remove(userId[0]);}// 自定义一个notification事件，也可以自定义其它任何名字的事件@OnEvent("notification")public void notification(String userId,Map<String, String> map) {if(socketMap.get(userId) != null)((SocketIOClient)socketMap.get(userId)).sendEvent("notification", map);}
}

4.配置启动类：

SocketServerRunner.java

import com.corundumstudio.socketio.SocketIOServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;@Component
@Order(1)
public class SocketServerRunner implements CommandLineRunner {@Autowiredprivate SocketIOServer server;@Overridepublic void run(String... args) {server.start();}
}

配置完成。

发送类：

@Autowired
private SocketEventHandler socketEventHandler;
Map<String,String> map = new HashMap<>();
map.put("type",police.getPoliceType());
map.put("title",police.getStationName()+"发生:"+police.getPoliceType());
map.put("text",police.getPoliceDescribe());
map.put("id",police.getId()+"");
String stationId = police.getStationId();
//查询有该权限的用户id
List<TbAdmin> tbAdmins = tbAdminMapper.selectIdByStationId(stationId);
socketEventHandler.notification("1",map);
socketEventHandler.notification("2",map);
if(null != tbAdmins && !tbAdmins.isEmpty()){for(TbAdmin admin : tbAdmins){socketEventHandler.notification(admin.getId().toString(),map);}
}

前端js：

var socket;
connect();function connect() {socket = io.connect('http://localhost:8082', {query: 'userId=' + $("#userId").val()});socket.on('connect', function () {console.log("连接成功");});//接收数据socket.on('notification', function (data) {var type = '';if(data.type == '高温报警'){type = 'error';}else{type = 'warn';}narn(type,data.title,data.text,data.id);});function narn (type,title,text,policeId) {naranja()[type]({title: title,text: text,timeout: 1000*60,buttons: [{text: '标记已读',click: function (e) {//改变状态changePoliceReadStatus(policeId);}}]})}
}
function changePoliceReadStatus(policeId){$.ajax({type:'post',url:ctx+'/police/changePoliceReadStatus?id='+policeId,})
}