From d7ee12cb208473741f79b64505adccd0653e4654 Mon Sep 17 00:00:00 2001 From: jinpengyong <jpy123456> Date: Tue, 27 Oct 2020 13:44:28 +0800 Subject: [PATCH] vue页面增加误差率参数,沧州剔除非本公司设备显示,报表接口更新 --- src/main/java/com/moral/webSocketServer/WebSocketServer.java | 43 ++++++++++++++++++++++--------------------- 1 files changed, 22 insertions(+), 21 deletions(-) diff --git a/src/main/java/com/moral/webSocketServer/WebSocketServer.java b/src/main/java/com/moral/webSocketServer/WebSocketServer.java index bdbd695..a6654ed 100644 --- a/src/main/java/com/moral/webSocketServer/WebSocketServer.java +++ b/src/main/java/com/moral/webSocketServer/WebSocketServer.java @@ -17,12 +17,18 @@ import com.moral.common.util.ParameterUtils; import com.moral.entity.Device; import com.moral.service.DeviceService; +import com.moral.util.RabbitMQUtils; +import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; +import com.rabbitmq.client.Consumer; +import com.rabbitmq.client.DefaultConsumer; +import com.rabbitmq.client.Envelope; import com.rabbitmq.client.QueueingConsumer; import lombok.extern.slf4j.Slf4j; + import org.springframework.stereotype.Component; /** @@ -62,45 +68,40 @@ this.session = session; //������������������������������������������������ webSocketSet.add(this); + int flag = param.indexOf("&"); int regionCodeIndex = param.indexOf("_"); orgId = param.substring(0, flag); - accountId = param.substring(flag + 1,regionCodeIndex); + accountId = param.substring(flag + 1, regionCodeIndex); regionCode = param.substring(regionCodeIndex + 1); String QUEUE_NAME = "deviceInfo_" + accountId; + Map<String, Object> paramMap = new HashMap<String, Object>(); paramMap.put("orgId", orgId); paramMap.put("regionCode", regionCode); ParameterUtils.getRegionType4RegionCode(paramMap); List<Device> deviceList = deviceService.queryDevice(paramMap); + + try { - //������������������������������������������������ - ConnectionFactory factory = new ConnectionFactory(); - //������MabbitMQ������������ip��������������� - factory.setHost("47.96.15.25"); - factory.setPort(5672); - factory.setUsername("guest"); - factory.setPassword("guest_pass"); String routingKey; - connection = factory.newConnection(); - channel = connection.createChannel(); + this.connection = RabbitMQUtils.getConnection(); + channel = this.connection.createChannel(); //��������������������������������������������������������������������������������������������������������� channel.queueDeclare(QUEUE_NAME, false, false, true, null); for (Device d : deviceList) { routingKey = orgId + "." + d.getMac(); channel.queueBind(QUEUE_NAME, "screens_data", routingKey); } - //��������������������� - QueueingConsumer consumer = new QueueingConsumer(channel); - //������������������ - channel.basicConsume(QUEUE_NAME, true, consumer); - - while (true) { - //nextDelivery������������������������������������������������������������take��������� - QueueingConsumer.Delivery delivery = consumer.nextDelivery(); - String message = new String(delivery.getBody()); - sendMessage(message); - } + channel.basicQos(30);//��������������� + channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel){ + @Override + public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { + String msg = new String(body, "utf-8"); + sendMessage(msg); + channel.basicAck(envelope.getDeliveryTag(), true); + } + }); } catch (Exception e) { log.error(e.getMessage()); } -- Gitblit v1.8.0