From 2257769c825d3451d4256f5a3e62421f08994557 Mon Sep 17 00:00:00 2001 From: luozukai Date: Mon, 4 Mar 2019 15:53:25 +0800 Subject: [PATCH] =?UTF-8?q?[=E9=93=9C=E9=94=A3=E6=B9=BE][=E4=BF=AE?= =?UTF-8?q?=E6=94=B9][=E9=A2=84=E8=B4=A6=E5=8D=95]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/WxFlowAbleController.java | 54 +++++-------------- .../iformall/mq/impl/RocketMqConsumer.java | 19 +++---- .../mq/impl/RocketMqMessageProducer.java | 49 ++++------------- 3 files changed, 30 insertions(+), 92 deletions(-) diff --git a/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java b/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java index 626db610c..823a6ca53 100644 --- a/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java +++ b/mallinkAdmin/src/main/java/com/iformall/controller/WxFlowAbleController.java @@ -1,16 +1,12 @@ package com.iformall.controller; import com.iformall.common.ResultData; -import com.iformall.domain.po.WxFlowRecord; -import com.iformall.domain.po.WxMsgRecord; -import com.iformall.enums.EnumFlowRecordStatus; -import com.iformall.mq.MqBaseProducer; +import com.iformall.mq.MqProducer; import com.iformall.service.WxFlowService; import io.swagger.annotations.Api; import io.swagger.annotations.ApiImplicitParam; import io.swagger.annotations.ApiImplicitParams; import io.swagger.annotations.ApiOperation; -import org.flowable.engine.RuntimeService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -27,7 +23,14 @@ public class WxFlowAbleController extends BaseController { @Autowired private WxFlowService wxFlowService; @Autowired - RuntimeService runtimeService; + private MqProducer mqProducer; + + @ApiOperation(value = "mqSend",notes = "") + @PostMapping("/mqSend") + public ResultData mqSend() throws Throwable{ + mqProducer.sendMessage("aaaa","topic-1","tag-1","key1"); + return new ResultData(); + } @ApiOperation(value = "启动流程",notes = "{\"businessId\":\"221178539888607232\",\"businessType\":1,\"remark\":\"意见意见。\",\"taskAssignee\":[{\"taskKey\":\"firstTaskUser\",\"assignee\":\"243650055783841792\"},{\"taskKey\":\"secondTaskUser\",\"assignee\":\"\"}],\"variables\":[{\"key\":\"contractType\",\"value\":\"1\"},{\"key\":\"contractNumber\",\"value\":\"1111\"}]}") @PostMapping("/start") @@ -36,12 +39,6 @@ public class WxFlowAbleController extends BaseController { return wxFlowService.start(params,super.getUser().getId(),super.getUser().getName(),super.getTenantId()); } - @GetMapping(value = "/del") - public ResultData list(String id) { - runtimeService.deleteProcessInstance(id, "驳回"); - return new ResultData(); - } - /** * 用户代办列表 */ @@ -50,30 +47,13 @@ public class WxFlowAbleController extends BaseController { @ApiImplicitParams({ @ApiImplicitParam(name = "pageNum", value = "页数", dataType = "int", paramType = "query", required = true), @ApiImplicitParam(name = "pageSize", value = "每页条数", dataType = "int", paramType = "query", required = true), - @ApiImplicitParam(name = "flowType", value = "流程类型,1合同流程 2账单流程", dataType = "string", paramType = "query", required = true) + @ApiImplicitParam(name = "flowType", value = "流程类型,1合同流程 2账单流程", dataType = "int", paramType = "query", required = true) }) - public ResultData list(String flowType,Integer pageNum, Integer pageSize) { + public ResultData list(Integer flowType,Integer pageNum, Integer pageSize) { logger.debug("[" + getIpAddr() + "] FlowAbleController::list"); return wxFlowService.list(flowType,pageNum,pageSize,super.getUser().getId()); } - /** - * 我的申请列表 - */ - @ApiOperation("我的申请列表") - @GetMapping(value = "/myApplyList") - @ApiImplicitParams({ - @ApiImplicitParam(name = "pageNum", value = "页数", dataType = "int", paramType = "query", required = true), - @ApiImplicitParam(name = "pageSize", value = "每页条数", dataType = "int", paramType = "query", required = true), - }) - public ResultData myApplyList(@ModelAttribute WxFlowRecord record, Integer pageNum, Integer pageSize) { - logger.debug("[" + getIpAddr() + "] FlowAbleController::myApplyList"); - record.setUserId(super.getUser().getId()); - record.setStatus(EnumFlowRecordStatus.NEW.getCode()); - return new ResultData(wxFlowService.listAsPage(record,pageNum,pageSize)); - } - - /** * 审批历史 */ @@ -114,20 +94,10 @@ public class WxFlowAbleController extends BaseController { */ @ApiOperation(value = "驳回",notes = "{\"taskId\":\"\",\"processInstanceId\":\"\",\"remark\":\"\"}") @PostMapping(value = "reject") - public ResultData reject(@RequestBody Map params) { + public ResultData reject(@RequestBody Map params) { return wxFlowService.reject(params,super.getUser().getId(),super.getUser().getName(),super.getTenantId()); } - - /** - * 撤回 - */ - @ApiOperation(value = "撤回",notes = "{\"taskId\":\"\",\"processInstanceId\":\"\",\"remark\":\"\"}") - @PostMapping(value = "setBack") - public ResultData setBack(@RequestBody Map params) { - return wxFlowService.setBack(params,super.getUser().getId(),super.getUser().getName()); - } - /** * 生成流程图 * diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java index 5c9bf0ff1..4d4e35442 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java @@ -1,23 +1,19 @@ -package com.iformall.mq.impl; +package com.iformall.mq; -import com.iformall.mq.MqBaseConsumer; -import com.iformall.mq.MQConfig; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.common.message.MessageExt; import org.rocketmq.starter.annotation.RocketMQListener; import org.rocketmq.starter.annotation.RocketMQMessage; -import org.springframework.context.annotation.Profile; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; @Slf4j -@Profile(MQConfig.Impl.ROCKET_MQ) +@Service @RocketMQListener(topic = "topic-1") -public class RocketMqConsumer extends MqBaseConsumer { +public class MqConsumer { + private final Logger log = LoggerFactory.getLogger(this.getClass()); -<<<<<<< HEAD:mallinkService/src/main/java/com/iformall/mq/impl/RocketMqConsumer.java - @RocketMQMessage(messageClass = String.class, tag = "tag-1") - public void onMessage(String message) { - doMessage(message); -======= @RocketMQMessage(messageClass = MessageExt.class, tag = "tag-1") public void onMessage(MessageExt message) { try { @@ -31,6 +27,5 @@ public class RocketMqConsumer extends MqBaseConsumer { e.printStackTrace(); } // return ConsumeConcurrentlyStatus.RECONSUME_LATER; ->>>>>>> [铜锣湾][修改][预账单]:mallinkService/src/main/java/com/iformall/mq/MqConsumer.java } } diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java index f34de661d..bb8c68801 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/RocketMqMessageProducer.java @@ -1,14 +1,5 @@ -package com.iformall.mq.impl; +package com.iformall.mq; -import com.iformall.domain.po.BaseMsg; -import com.iformall.domain.po.WxMsg; -import com.iformall.domain.po.WxMsgRecord; -import com.iformall.enums.EnumMsgRecordStatus; -import com.iformall.mapper.WxMsgRecordMapper; -import com.iformall.mq.MQConfig; -import com.iformall.mq.MqBaseProducer; -import com.iformall.service.WxMsgRecordService; -import com.iformall.utils.JsonUtil; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendCallback; @@ -17,39 +8,30 @@ import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Service; + import javax.annotation.PostConstruct; -import java.util.UUID; /** - * @Auther: Stormeye + * @Auther: luozukai * @Date: 2019-01-31 * @Description: RocketMQ消息生产者 */ @Service @Slf4j -@Profile(MQConfig.Impl.ROCKET_MQ) -public class RocketMqMessageProducer implements MqBaseProducer { +public class MqProducer { private final Logger log = LoggerFactory.getLogger(this.getClass()); - + private static final DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer"); @Value("${spring.rocketmq.nameServer}") private String namesrvAddr; - private static final DefaultMQProducer producer = new DefaultMQProducer("DefaultProducer"); - - @Autowired - private WxMsgRecordMapper wxMsgRecordMapper; - - @Autowired - private WxMsgRecordService wxMsgRecordService; - @PostConstruct public void start(){ try { producer.setNamesrvAddr(namesrvAddr); + //producer.setRetryTimesWhenSendFailed(5);//重试机制,默认2次 + //producer.setSendMsgTimeout(6000); //默认是3000 producer.start(); log.info("Message Producer Start..."); }catch (Exception e){ @@ -57,21 +39,14 @@ public class RocketMqMessageProducer implements MqBaseProducer { } } - @Override - public void sendMessage(BaseMsg data, String topic, String tags, String keys) { + public void sendMessage(String data, String topic, String tags, String keys) throws Throwable{ try { - wxMsgRecordService.save(data); - - byte[] messageBody = JsonUtil.obj2Json(data).getBytes(RemotingHelper.DEFAULT_CHARSET); - + byte[] messageBody = data.getBytes(RemotingHelper.DEFAULT_CHARSET); Message mqMsg = new Message(topic, tags, keys, messageBody); - producer.send(mqMsg, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { log.info("Message Producer: Send Message Success {}", sendResult); -// data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode()); -// wxMsgRecordMapper.update(data); } @Override @@ -79,10 +54,8 @@ public class RocketMqMessageProducer implements MqBaseProducer { log.error("Message Producer: Send Message Error ", throwable); } }); - } catch (Exception e) { - log.error("Message Producer: Send Message Error ", e); + } catch (Throwable e) { + e.printStackTrace(); } - } - }