后台应该框架使用的是springboot,必须的依赖如下:
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId></dependency><dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>4.1.0</version></dependency>
添加config文件:
//@Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter rverEndpointExporter() { return new ServerEndpointExporter(); }}
添加wobsocket相关文件类:
//该注解用来指定一个URI,客户端可以通过这个URI来连接到WebSocket。类似Servlet的注解mapping。无需在web.xml中配置。 @ServerEndpoint(value = "/websocket")@Componentpublic class MyWebSocket {@Value("${spring.rabbitmq.host:172.16.1.29}")private String rabbitHost;@Value("${spring.rabbitmq.port:5672}")private Integer rabbitPort;@Value("${spring.rabbitmq.urname:admin}")private String rabbitUr;@Value("${spring.rabbitmq.password:a123456}")private String rabbitPass;private static Logger log = LogManager.getLogger(MyWebSocket.class);//静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。 // @Autowired // public TerminalService terminalServiceInWebSocket; //与某个客户端的连接会话,需要通过它来给客户端发送数据 private Session ssion;private static CopyOnWriteArraySet<Session> ssions = new CopyOnWriteArraySet<Session>();private static int onlineCount = 0;//concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。若要实现服务端与单一客户端通信的话,可以使用Map来存放,其中Key可以为用户标识 private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();/** * 连接建立成功调用的方法 * * @param ssion 可选的参数。ssion为与某个客户端的连接会话,需要通过它来给客户端发送数据 */ @OnOpen public void onOpen(Session ssion) {this.ssion = ssion;ssions.add(ssion);webSocketSet.add(this); //加入t中 addOnlineCount(); //在线数加1 log.info("有新连接加入!当前在线人数为" + getOnlineCount());String QUEUE_NAME = "clodPlace";try {//创建连接连接到MabbitMQ ConnectionFactory factory = new ConnectionFactory();//设置MabbitMQ所在主机ip或者主机名 factory.tHost("172.16.1.29");// factory.tPort(rabbitPort); factory.tUrname("admin");factory.tPassword("a123456");//创建一个连接 Connection connection = factory.newConnection();//创建一个频道 Channel channel = connection.createChannel();//指定一个队列 /* durable:true、fal.true:在服务器重启时,能够存活 exclusive :是否为当前连接的专用队列,在连接断开后,会自动删除该队列,生产环境中应该很少用到吧。 autodelete:当没有任何消费者使用时,自动删除该队列。this means that the queue will be deleted when there are no more process consuming messages from it. */ channel.queueDeclare(QUEUE_NAME, true, fal, fal, null);//每次从队列获取的数量,保证一次只分发一个 channel.basicQos(1);log.info(" [*] Waiting for messages.");//监听队列 Consumer consumer = new DefaultConsumer(channel) {@Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//处理监听得到的消息 String message = null;try {message = new String(body, "UTF-8");//消息处理逻辑 ndMessage(message);} catch (UnsupportedEncodingException e) {e.printStackTrace();channel.abort();} finally {log.info("[x] Done.");channel.basicAck(envelope.getDeliveryTag(), fal);}log.info("[x] Received '" + message + "'");}};//autoAck是否自动回复,如果为true的话,每次生产者只要发送信息就会从内存中删除,那么如果消费者程序异常退出, // 那么就无法获取数据,我们当然是不希望出现这样的情况,所以才去手动回复,每当消费者收到并处理信息然后在通知 // 生成者。最后从队列中删除这条信息。如果消费者异常退出,如果还有其他消费6年级数学者,那么就会把队列中的消息发送给其 // 他消费者,如果没有,等消费者启动时候再次发送。 boolean autoAck = fal;//消息消费完成确认 channel.basicConsume(QUEUE_NAME, autoAck, consumer);} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}/** * 连接关闭调用的方法 */ @OnClo public void onClo() {webSocketSet.remove(this); //从t中删除 ssions.remove(ssion);subOnlineCount(); //在线数减1 log.info("有一连接关闭!当前在线人数为" + getOnlineCount());}/** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息 * @param ssion 可选的参数 */ @OnMessage public void onMessage(String message, Session ssion) {log.info("来自客户端的消息:" + message);//群发消息 for (MyWebSock对口升学考试et item : webSocketSet) {try {item.ndMessage(message);} catch (IOException e) {e.printStackTrace();continue;}}}/** * 发生错误时调用 * * @param ssion * @param error */ @OnError public void onError(Session ssion, Throwable error) {log.error("发生错误");error.printStackTrace();}/** * 这个方法与上面几个方法不一样。没有用注解,是根据自己需要添加的方法。 * * @param message * @throws IOException */ public void ndMessage(String message) throws IOException {//阻塞式的(同步的) if (ssions.size() != 0) {for (Session s : ssions) {if (s != null) {s.getBasicRemote().ndText(message);}}}//非阻塞式的(异步的) // this.ssion.getAsyncRemote().ndText(message); log.info("[x] 推送消息"+message);}public static synchronized int getOnlineCount() {return onlineCount;}public static synchronized void addOnlineCount() {MyWebSocket.onlineCount++;}public static synchronized void subOnlineCount() {MyWebSocket.onlineCount社会实践活动登记表--;}}
添加订阅配置:
@Configurationpublic class RabbitTopic {final static String message = "clodPlace";final static String messages = "topic.messages";//创建队列 @Bean public Queue queueMessage() {return new Queue(RabbitTopic.message);}//创建队列 @Bean public Queue queueMessages() {return new Queue(RabbitTopic.messages);}//创建交换器 @Bean TopicExchange exchange() {return new TopicExchange("topicExchange");}//对列绑定并关联到ROUTINGKEY @Bean Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {return BindingBuilder.bind(queueMessage).to(exchange).with(RabbitTopic.message);}//对列绑定并关联到ROUTINGKEY @Bean Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一个词,#表示零个或多个词 }
web的html文件:
<!DOCTYPE html><html> <head> <meta chart="UTF-8"><title>WebSocket</title> </head> <body> <div id="main"> <div id="message"></div> </div> </b雅思新东方ody&make短语gt; <script type="text/javascript"> var websocket = null; //判断当前浏览器是否支持WebSocket if ('WebSocket' in window) { websocket = new WebSocket("ws://192.168.2.59:8080/springboot/websocket"); } el { alert('当前浏览器 不支持WebSocket') } //连接发生错误的回调方法 websocket.onerror = function () { tMessageInnerHTML("连接发生错误"); }; //连接成功建立的回调方法 websocket.onopen = function () { tMessageInnerHTML("连接成功"); } //接收到消息的回调方法,此处添加处理接收消息方法,当前是将接收到的信息显示在网页上 websocket.onmessage = function (event) { tMessageInnerHTML(event.data); } //连接关闭的回调方法 websocket.onclo = function () { tMessageInnerHTML("连接关闭,如需登录请刷新页面。"); } //监听窗口关闭事件,当窗口关闭时,主动去关闭websocket连接,防止连接还没断开就关闭窗口,rver端会抛异常。 window.onbeforeunload = function () { cloWebSocket(); } //将消息显示在网页上,如果不需要显示在网页上,则不调用该方法 function tMessageInnerHTML(innerHTML) { document.getElementById('message').innerHTML += innerHTML + '<br/>'; } </script> </html>
本文发布于:2023-04-07 18:25:49,感谢您对本站的认可!
本文链接:https://www.wtabcd.cn/fanwen/zuowen/2225142e43974e4eb2c8f724db3e4289.html
版权声明:本站内容均来自互联网,仅供演示用,请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系,我们将在24小时内删除。
本文word下载地址:Rabbitmq+WebSocket实现消息推送给web前端 原创.doc
本文 PDF 下载地址:Rabbitmq+WebSocket实现消息推送给web前端 原创.pdf
留言与评论(共有 0 条评论) |