Commit 04b063d8 by 赵剑炜

消息队列日志监听

parent fc20c5ab
...@@ -4,7 +4,13 @@ import com.alibaba.fastjson.JSON; ...@@ -4,7 +4,13 @@ import com.alibaba.fastjson.JSON;
import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.CabinetLogsRecordReq; import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.CabinetLogsRecordReq;
import com.junmp.jyzb.api.bean.req.RabbitMqOrderReq; import com.junmp.jyzb.api.bean.req.RabbitMqOrderReq;
import com.junmp.jyzb.service.InventorySummaryService; import com.junmp.jyzb.service.InventorySummaryService;
import com.junmp.v2.log.api.LogRecordApi;
import com.junmp.v2.log.api.bean.dto.LogRecordDto;
import com.junmp.v2.log.api.constant.LogConstant;
import com.junmp.v2.log.api.enums.LogTypeEnum;
import com.junmp.v2.log.api.factory.LogRecordFactory;
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Channel;
import org.apache.poi.ss.formula.functions.Now;
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Header;
...@@ -12,6 +18,7 @@ import org.springframework.stereotype.Component; ...@@ -12,6 +18,7 @@ import org.springframework.stereotype.Component;
import javax.annotation.Resource; import javax.annotation.Resource;
import java.io.IOException; import java.io.IOException;
import java.sql.Time;
/** /**
* 单警柜出入库记录上报并且记录到数据库中 * 单警柜出入库记录上报并且记录到数据库中
...@@ -21,23 +28,36 @@ import java.io.IOException; ...@@ -21,23 +28,36 @@ import java.io.IOException;
public class CabinetLogReceiver { public class CabinetLogReceiver {
@Resource @Resource
private InventorySummaryService inventorySummaryService; private InventorySummaryService inventorySummaryService;
/**
* 日志记录的api
*/
@Resource
private LogRecordApi logRecordApi;
@RabbitListener(queues = "CabinetLogsRecord")//监听队列名 @RabbitListener(queues = "CabinetLogsRecord")//监听队列名
private void a(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){ private void a(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){
// 创建日志对象
LogRecordDto logRecordDTO = LogRecordFactory.createLogRecord(LogConstant.LOG_DEFAULT_NAME, "CabinetLogsRecord");
logRecordDTO.setLogType(LogTypeEnum.OPT_LOG.getCode());
logRecordDTO.setHttpMethod("RabbitMQ");
logRecordDTO.setLogContent("CabinetLogsRecord");
logRecordDTO.setRequestResult(msg);
try { try {
System.out.println("msg = " + msg);
// 手动确认消息 // 手动确认消息
channel.basicAck(deliveryTag, false); channel.basicAck(deliveryTag, false);
CabinetLogsRecordReq req = JSON.parseObject(msg, CabinetLogsRecordReq.class); CabinetLogsRecordReq req = JSON.parseObject(msg, CabinetLogsRecordReq.class);
//将消息进行消费 //将消息进行消费
Boolean result = inventorySummaryService.cabinetInOutRecords(req); Boolean result = inventorySummaryService.cabinetInOutRecords(req);
logRecordDTO.setRequestResult(result.toString());
logRecordApi.addAsync(logRecordDTO);
} catch (Exception e) { } catch (Exception e) {
// 处理消息过程中发生异常,拒绝消息并将其重新放回队列ms
try { try {
channel.basicNack(deliveryTag, false, true); channel.basicNack(deliveryTag, false, true);
logRecordDTO.setRequestResult(e.getMessage().toString());
logRecordApi.addAsync(logRecordDTO);
} catch (IOException ex) { } catch (IOException ex) {
logRecordDTO.setRequestResult(ex.getMessage().toString());
logRecordApi.addAsync(logRecordDTO);
throw new RuntimeException(ex); throw new RuntimeException(ex);
} }
} }
......
...@@ -4,6 +4,11 @@ import com.alibaba.fastjson.JSON; ...@@ -4,6 +4,11 @@ import com.alibaba.fastjson.JSON;
import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.CabinetLogsRecordReq; import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.CabinetLogsRecordReq;
import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.OutInLogsReq; import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.OutInLogsReq;
import com.junmp.jyzb.service.LogSummaryService; import com.junmp.jyzb.service.LogSummaryService;
import com.junmp.v2.log.api.LogRecordApi;
import com.junmp.v2.log.api.bean.dto.LogRecordDto;
import com.junmp.v2.log.api.constant.LogConstant;
import com.junmp.v2.log.api.enums.LogTypeEnum;
import com.junmp.v2.log.api.factory.LogRecordFactory;
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.AmqpHeaders;
...@@ -21,23 +26,39 @@ public class OutInLogsByCabinetReceiver { ...@@ -21,23 +26,39 @@ public class OutInLogsByCabinetReceiver {
@Resource @Resource
private LogSummaryService logSummaryService; private LogSummaryService logSummaryService;
/**
* 日志记录的api
*/
@Resource
private LogRecordApi logRecordApi;
@RabbitListener(queues = "CabinetLogsRecord")//监听队列名 @RabbitListener(queues = "CabinetLogsRecord")//监听队列名
private void a(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){ private void a(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){
// 创建日志对象
LogRecordDto logRecordDTO = LogRecordFactory.createLogRecord(LogConstant.LOG_DEFAULT_NAME, "CabinetLogsRecord");
logRecordDTO.setLogType(LogTypeEnum.OPT_LOG.getCode());
logRecordDTO.setHttpMethod("RabbitMQ");
logRecordDTO.setLogContent("CabinetLogsRecord");
logRecordDTO.setRequestResult(msg);
try { try {
System.out.println("msg = " + msg); System.out.println("msg = " + msg);
// 手动确认消息 // 手动确认消息
channel.basicAck(deliveryTag, false); channel.basicAck(deliveryTag, false);
CabinetLogsRecordReq req = JSON.parseObject(msg, CabinetLogsRecordReq.class); CabinetLogsRecordReq req = JSON.parseObject(msg, CabinetLogsRecordReq.class);
//将消息进行消费 //将消息进行消费
Boolean result = logSummaryService.processOutInRecords(req); Boolean result = logSummaryService.processOutInRecords(req);
logRecordDTO.setRequestResult(result.toString());
logRecordApi.addAsync(logRecordDTO);
} catch (Exception e) { } catch (Exception e) {
// 处理消息过程中发生异常,拒绝消息并将其重新放回队列ms // 处理消息过程中发生异常,拒绝消息并将其重新放回队列ms
try { try {
channel.basicNack(deliveryTag, false, true); channel.basicNack(deliveryTag, false, true);
logRecordDTO.setRequestResult(e.getMessage().toString());
logRecordApi.addAsync(logRecordDTO);
} catch (IOException ex) { } catch (IOException ex) {
logRecordDTO.setRequestResult(ex.getMessage().toString());
logRecordApi.addAsync(logRecordDTO);
throw new RuntimeException(ex); throw new RuntimeException(ex);
} }
} }
......
...@@ -4,6 +4,11 @@ import com.alibaba.fastjson.JSON; ...@@ -4,6 +4,11 @@ import com.alibaba.fastjson.JSON;
import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.OutInLogsReq; import com.junmp.jyzb.api.bean.query.InAndOutRecordReq.OutInLogsReq;
import com.junmp.jyzb.service.LogSummaryService; import com.junmp.jyzb.service.LogSummaryService;
import com.junmp.jyzb.service.OrderMainService; import com.junmp.jyzb.service.OrderMainService;
import com.junmp.v2.log.api.LogRecordApi;
import com.junmp.v2.log.api.bean.dto.LogRecordDto;
import com.junmp.v2.log.api.constant.LogConstant;
import com.junmp.v2.log.api.enums.LogTypeEnum;
import com.junmp.v2.log.api.factory.LogRecordFactory;
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.AmqpHeaders;
...@@ -16,28 +21,47 @@ import java.io.IOException; ...@@ -16,28 +21,47 @@ import java.io.IOException;
/** /**
* 本地主机上报出入记录监听消息队列 * 本地主机上报出入记录监听消息队列
*/ */
@Component @Component
public class OutInLogsReceiver { public class OutInLogsReceiver {
@Resource @Resource
private LogSummaryService logSummaryService; private LogSummaryService logSummaryService;
/**
* 日志记录的api
*/
@Resource
private LogRecordApi logRecordApi;
@RabbitListener(queues = "11111")//监听队列名 @RabbitListener(queues = "11111")//监听队列名
private void a(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){ private void a(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){
LogRecordDto logRecordDTO = LogRecordFactory.createLogRecord(LogConstant.LOG_DEFAULT_NAME, "CabinetLogsRecord");
logRecordDTO.setLogType(LogTypeEnum.OPT_LOG.getCode());
logRecordDTO.setHttpMethod("RabbitMQ");
logRecordDTO.setLogContent("OrderResult");
logRecordDTO.setRequestResult(msg);
try { try {
System.out.println("msg = " + msg); System.out.println("msg = " + msg);
// 创建日志对象
// 手动确认消息 // 手动确认消息
channel.basicAck(deliveryTag, false); channel.basicAck(deliveryTag, false);
OutInLogsReq req = JSON.parseObject(msg, OutInLogsReq.class); OutInLogsReq req = JSON.parseObject(msg, OutInLogsReq.class);
//将消息进行消费 //将消息进行消费
Boolean result = logSummaryService.processInventoryRecords(req); Boolean result = logSummaryService.processInventoryRecords(req);
logRecordDTO.setRequestResult(result.toString());
logRecordApi.addAsync(logRecordDTO);
} catch (Exception e) { } catch (Exception e) {
// 处理消息过程中发生异常,拒绝消息并将其重新放回队列ms // 处理消息过程中发生异常,拒绝消息并将其重新放回队列ms
try { try {
channel.basicNack(deliveryTag, false, true); channel.basicNack(deliveryTag, false, true);
logRecordDTO.setRequestResult(e.getMessage().toString());
logRecordApi.addAsync(logRecordDTO);
} catch (IOException ex) { } catch (IOException ex) {
logRecordDTO.setRequestResult(ex.getMessage().toString());
logRecordApi.addAsync(logRecordDTO);
throw new RuntimeException(ex); throw new RuntimeException(ex);
} }
} }
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论