配置文件:
pom文件:
<spring-rabbit.version>1.7.2.RELEASE</spring-rabbit.version>
<amqp-client.version>3.6.1</amqp-client.version>
<rabbit.mq.address></rabbit.mq.address>
<rabbit.mq.userName>guest</rabbit.mq.userName>
<rabbit.mq.password>guest</rabbit.mq.password>
Properties文件,properties引用pom中的配置:
wfpt.properties:
wfpt.rabbitmq.address=${rabbit.mq.address}
wfpt.rabbitmq.userName=${rabbit.mq.userName}
wfpt.rabbitmq.password=${rabbit.mq.password}
wfpt.rabbitmq.virtualHost=${rabbit.mq.virtualHost}
RabbiitMQ使用的模式:
工作流中使用的是Routingkey 模式,但是没有指定Exchange的类型和名称,当Exchange没有指定类型和名称时,RabbitMQ将会为每一个队列设定一个Default Exchange,它的Routing key为消息队列的名称。见下图:
RabbitMQ生产者:ActRabbitMQProducer
RabbitMQ消费者:ActRabbitMQConsumer
控制是否推送代办:(发起流程时,是否推送待办)
IActPushTODOService中的isPushTODO(DelegateTask delegateTask)方法,该方法根据流程变量中是否存在isPushTODO变量来决定是否推送待办。
RabbitMQ与KafKa的切换:(状态模式的使用)
1. 控制使用的消息中间件的类型:ActTodoTaskMQManager,根据配置文件中wfpt.todoTask.sendType的配置的类型和数据库中act_ex_system表的pushType的值来判断使用的消息中间件是哪个。
2. 状态接口ActTodoTaskMQState,RabbitMQ和KafKa发送方法共同实现该接口。
3. ActRabbitMQTodoTaskService和ActKafkaTodoTaskService方法时发送代办的实现类
4. 实际调用形式:
1. ActTodoTaskMQManager mqManager = new ActTodoTaskMQManager();
2. ActTodoTaskMQState state = mqManager.getMQState();
3. state.sendTodoMsgAndUpdateTodoTask(distributeTaskDTO,todoTaskList);
系统说明:
(1)pom文件依赖:
<!-- RabbitMQ -->
<spring-rabbit.version>1.7.2.RELEASE</spring-rabbit.version>
<amqp-client.version>3.6.1</amqp-client.version>
<!-- 工作流服务(开发环境) -->
<wfpt.version>0.0.5-SNAPSHOT</wfpt.version>
<!-- RabbitMQ start -->
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>${spring-rabbit.version}</version>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>${amqp-client.version}</version>
</dependency>
<!-- RabbitMQ end -->
<!-- 工作流服务wfpt start -->
<dependency>
<groupId>com.yixin</groupId>
<artifactId>wfpt-rpc</artifactId>
<version>${wfpt.version}</version>
</dependency>
<!-- 工作流服务wfpt end -->
(2)xml配置文件:rpc-impl层,spring-context.xml
1. <!-- RabbitMQ接收待办:请注意默认是不启用状态,否则有可能产生消费了别人待办的消息 -->
<bean id="ActRabbitReceiveTodoTaskMessage"
class="com.yixin.wfpt.api.todotask.rabbitmq.ActRabbitReceiveTodoTaskMessage" init-method="init">
<property name="queueName" value="sample"/>
<property name="initFlag" value="true"/>
<property name="receiveClass" value="com.yixin.demo.todotask.ActExDemoTodoTaskServiceImpl"/>
</bean>
说明:ActRabbitReceiveTodoTaskMessage类是上面所讲的接收待办的客户端,工作流deploy的时候会把该类deploy到nexus,sample会依赖到该类
ActExDemoTodoTaskServiceImpl类是业务系统接收待办的实现类,该类由业务系统实现,该类实现了接口IActExDemoTodoTaskService,该接口是工作流提供的接口
- 接收待办的客户端
ActRabbitReceiveTodoTaskMessage类
public void init(){
if(initFlag){
logger.info("接收待办,连接RabbitMQ开始!");
if(!StringUtils.hasText(queueName)){
logger.info("RabbitMQ接收待办消息,传入参数queueName为空!");
return ;
}
logger.info("RabbitMQ接收待办消息,queueName为:{}",queueName);
if (queueName.endsWith(",")) {
logger.info("RabbitMQ接收待办消息,queueName以逗号结尾,不符合规范,queueName={}",queueName);
return;
}
if(queueName.indexOf(',') > -1){
String[] queues = queueName.split(",");
for(String queue : queues){
consumeMsg(queue);
}
}else{
consumeMsg(queueName);
}
}
}
说明:
(1) init方法是在业务系统spring-context.xml文件中配置的init-method="init",在业务系统启动的时候而且是initFlag为true的时候执行该方法。
(2) 支持多个队列,以逗号分隔。queueName指的是接入系统的名称,该出配置的queueName需要与工作流页面中接入系统的编码一样。
private void consumeMsg(String queue){
try {
/**
* 创建channel
*/
final Channel channel= ActRabbitMQConsumer.createChannel();
/**
* 声明队列
*/
channel.queueDeclare(queue,true,false,false,null);
channel.basicConsume(queue,false,newQueueingConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] bodys) throws IOException{
String message = new String(bodys);
List<TodoTaskDTO> todoTaskList = parseMessage(message);
execute(todoTaskList);
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}catch(Exception e) {
logger.error("RabbitMQ消费者接收消息失败!", e);
throw new BzException("RabbitMQ消费者接收消息失败!");
}
}
说明:该方法用于接收消息,basicConsume(queue,false,new QueueingConsumer())中第二个参数表示是否自动确认,此处为false表示手工确认。手工确认的方法为:channel.basicAck(envelope.getDeliveryTag(),false);
handleDelivery方法是监听器监听消息,如果RabbitMQ中存在消息则接收消息。
execute方法用applicationContext.getBean的方式,实现IActExDemoTodoTaskService接口的方法,receiveClass就是现实IActExDemoTodoTaskService接口的方法。
IPushTodoTaskAPI接口:
该接口时工作流提供的接口,但是由业务系统去实现该接口。