我家小花儿

这个屌丝很懒，什么也没留下！

热门标签

Spring Boot 整合rabbitmq_springboot整合rabbitmq

作者：我家小花儿 | 2024-04-17 06:48:34

踩

springboot整合rabbitmq

一、利用rabbitmq客户端手写生产者和消费者

目录结构以及需要的类:

1. 在pom文件里添加依赖rabbitmq client:


    <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>3.6.5</version>
    </dependency>

2. 新建connection工具类，设置虚拟主机、端口、用户名和密码，默认端口为5672


package com.example.shop.common.utils.rabbitmq;
 
 
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class ConnectionUtil {
 
 
    private static final String host = "localhost";
    private static final Integer port = 5672;
    private static final String username = "guest";
    private static final String password = "guest";
    private static final String visualHost = "/";
 
    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(visualHost);
        return connectionFactory.newConnection();
    }
 
}

3. 新建消息生产者工具类


package com.example.shop.common.utils.rabbitmq;
 
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class ProducerUtil {
 
    private static Connection connection;
 
 
    // 生产者是产生消息
    public static void produceMessage(String message) throws IOException, TimeoutException {
        connection = ConnectionUtil.getConnection();
        // 1. 创建一个信道 channel
        Channel channel = connection.createChannel();
        // 2. 声明一个交换机
        channel.exchangeDeclare("goods-exchange", "direct", true);
        // 3. 创建一个队列
        channel.queueDeclare("bug-goods", true, false, false, null);
        // 4. 交换机绑定队列, 定义routing key为good001
        channel.queueBind("bug-goods", "goods-exchange", "good001");
        // 5. 发布消息
        channel.basicPublish("goods-exchange", "good001", null, message.getBytes());
        System.out.println("消息发布成功...");
    }
 
}


下面是对上面几个方法的用法解析:
channel.exchangeDeclare("goods-exchange", "direct", true) :  声明一个交换机，名称为goods-exchange, 交互机的作用是后面可以绑定消息队列，然后通过匹配routing key来指定路由到哪个队列。 arg1 : 交换机名称, arg2 : 交换机类型(direct、topic、fanout), arg3: durable, 是否持久化。
channel.queueDeclare("bug-goods", true, false, false, null): 声明一个队列，名称为 bug-goods, 查看源码的实现类，可得知几个参数的含义: arg1: 消息队列名称， arg2: 是否持久化, arg3: exclusive, 是否私有，只能被第一个消费者消费，消费后其他消费者不能再消费到, arg4: 是否自动删除消息, arg5: 一些参数。

 this.delegate.queueDeclare(queue, durable, exclusive, autoDelete, arguments);

channel.queueBind("bug-goods", "goods-exchange", "good001");  将对列绑定指定的exchange, 并指定routing key ,如果 声明的exchange 的type 为fanout ，那么不需要指定routing key。 arg1: 队列名称， arg2: 交换机名称, arg3: routing key。channel.basicPublish("goods-exchange", "good001", null, message.getBytes());  往指定交换机发布一条消息。 arg1: 交换机名称，arg2: routing key, 如果没有这个参数是不可以的，如果不指定该参数那么会不指定路由到哪个队列，像是被阻塞了一样。 arg3: 基础属性， arg4: 消息内容，以字节形式传输。

4. 新建消费者工具类

通过basicConsume() 方法指定rabbitmq客户端监听消息队列的名称，在handleDelivery()方法里接收消息。

需要注意的是: 客户端需要向rabbitmq 服务确认已经接收到消息，存在rabbitmq里的消息才会删除掉。


package com.example.shop.common.utils.rabbitmq;
 
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class ConsumerUtil {
 
    private static Connection connection;
 
    // 消费者接收消息
    public static void reciveMsg() throws IOException, TimeoutException {
        connection = ConnectionUtil.getConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 创建消费者,绑定消息队列, 第二个参数为自动确认
        channel.basicConsume("bug-goods", true, new Consumer() {
            @Override
            public void handleConsumeOk(String s) {
                System.out.println("handleConsumeOk:");
            }
 
            @Override
            public void handleCancelOk(String s) {
                System.out.println("handleCancelOk:" + s);
 
            }
 
            @Override
            public void handleCancel(String s) throws IOException {
                System.out.println("handleCancel:" + s);
 
            }
 
            @Override
            public void handleShutdownSignal(String s, ShutdownSignalException e) {
                System.out.println("handleShutdownSignal:" + s);
 
            }
 
            @Override
            public void handleRecoverOk(String s) {
                System.out.println("handleRecoverOk:" + s);
 
 
            }
 
            @Override
            public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                System.out.println("handleDelivery:" + s);
                String msg = new String(bytes);
                System.out.println("收到消息:" + msg);
                // 可以根据rounting key来判断是不是自己想要的消息。在微服务架构中，可以使用服务名作为rounting key,实现单一接收消息。
                System.out.println("routing key: " + envelope.getRoutingKey());
                System.out.println("exchange: " + envelope.getExchange());
            }
        });
    }
}

5. 测试


package com.example.shop.common.utils.rabbitmq;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class MessageTest {
 
 
    public static void main(String[] args) throws IOException, TimeoutException {
 
        // 1. 发布一条消息
        ProducerUtil.produceMessage("您好!");
 
        // 2. 接收消息
        ConsumerUtil.reciveMsg();
 
    }
}

执行main方法后，会在队列里自动创建exchange和queue。

可以发现此时还是闲置状态。

场景一:

不确认消息接收，执行一次Main()方法，观察queue队列里的消息。

可以发现有一条消息unacked，就是rabbitmq不知道消费方有没有接收到消息。

场景二:

开启自动确认接收消息，确认接收消息，观察控制台和queue队列里的消息。

可以发现控制台收到了两条消息，队列的消息也被清掉，由此可以发现，rabbitmq只有确认接收到消息后才会清掉队列里的消息。

二、Spring Boot 整合rabbitmq

一、在management 桌面配置虚拟主机、交换机和队列

1. 先进入到rabbitmq的管理页面，然后在管理页面新建一个虚拟主机，默认的虚拟主机为"/" ,不同的环境使用不同的虚拟主机，如开发环境可以设置为dev,生产环境可以设置为prod。

2. 在默认的虚拟主机下，新增一个队列rabbitmqtest

二、测试案例

1. 添加rabbitmq依赖：


<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
 </dependency>

2.如果下载不下来，在settings.xml文件中添加以下镜像:


 <mirror>
      <id>nexus-aliyun</id>
      <name>Nexus aliyun</name>
      <mirrorOf>central</mirrorOf> 
      <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</mirror>
	
<mirror>
      <id>nexus</id>
       <name>internal nexus repository</name>
      <!-- <url>http://192.168.1.100:8081/nexus/content/groups/public/</url>-->
      <url>http://repo.maven.apache.org/maven2</url>
       <mirrorOf>central</mirrorOf>
 </mirror>

3. 新建一个controller,代码如下:


package com.example.rabbitmq.web;
 
 
import org.junit.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
@RestController
@RequestMapping("/api/user")
public class UserController {
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Autowired
    private Getmsg getmsg;
 
 
    @GetMapping("/send/msg")
    public String sendmsgtormq(){
        //指定向队列名为rabbitmqtest的消息队列中发送消息
        rabbitTemplate.convertAndSend("rabbitmqtest","发送一条消息");
       return  "消息发送成功!";
    }
 
 
 
}

4. 新建一个rabbitmq的监听器类，GetMsg，代码如下


package com.example.rabbitmq.web;
 
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@RabbitListener(queues = "rabbitmqtest")
@Component
public class Getmsg {
 
    @RabbitHandler
    public void getmsg(String string) {
        System.out.println("ranbbitmqtest接收到的消息为:"+string);
    }
}

5. 启动项目，输入地址: http://localhost:9000/api/user/send/msg ,监听器类收到队列的消息，队列的消息被消费后，默认会清除该条消息。

6. 另外附一个常见问题，如果在启动的时候报错:
aused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue 'rabbitmqtest' in vhost '/dev', class-id=50, method-id=10)

请检查该虚拟主机下是否存在有该名字对应的队列。

三、 rabbitmq开发实战

1. 项目中需要将提交的单据到工作流，然后工作流会给前端一个响应为提交成功或者失败。业务模块会根据提交的工作流来更新对应的单据状态并根据单据的状态来做一些业务逻辑。

发送消息代码用到了一个ApplicationEventPublisher 里的一个 publishEvent 方法，参数为object，支持任意类型的消息内容。


/*
 * Copyright 2002-2016 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
package org.springframework.context;
 
/**
 * Interface that encapsulates event publication functionality.
 * Serves as super-interface for {@link ApplicationContext}.
 *
 * @author Juergen Hoeller
 * @author Stephane Nicoll
 * @since 1.1.1
 * @see ApplicationContext
 * @see ApplicationEventPublisherAware
 * @see org.springframework.context.ApplicationEvent
 * @see org.springframework.context.event.EventPublicationInterceptor
 */
@FunctionalInterface
public interface ApplicationEventPublisher {
 
	/**
	 * Notify all <strong>matching</strong> listeners registered with this
	 * application of an application event. Events may be framework events
	 * (such as RequestHandledEvent) or application-specific events.
	 * @param event the event to publish
	 * @see org.springframework.web.context.support.RequestHandledEvent
	 */
	default void publishEvent(ApplicationEvent event) {
		publishEvent((Object) event);
	}
 
	/**
	 * Notify all <strong>matching</strong> listeners registered with this
	 * application of an event.
	 * <p>If the specified {@code event} is not an {@link ApplicationEvent},
	 * it is wrapped in a {@link PayloadApplicationEvent}.
	 * @param event the event to publish
	 * @since 4.2
	 * @see PayloadApplicationEvent
	 */
	void publishEvent(Object event);
 
}

ApplicationEventPublisher 类在 spring-context包里:

组装生产者消息:


   protected WorkflowCustomRemoteEvent createWorkflowCustomRemoteEvent(WorkFlowDocumentRef workFlowDocumentRef) {
        Assert.notNull(workFlowDocumentRef, "workFlowDocumentRef null");
 
        WorkflowMessageCO workflowMessageCO = new WorkflowMessageCO();
        workflowMessageCO.setUserBean(OrgInformationUtil.getUser());
        workflowMessageCO.setEntityOid(workFlowDocumentRef.getDocumentOid());
        workflowMessageCO.setEntityType(workFlowDocumentRef.getDocumentCategory().toString());
        workflowMessageCO.setStatus(workFlowDocumentRef.getStatus());
        workflowMessageCO.setUserId(workFlowDocumentRef.getCreatedBy());
        workflowMessageCO.setDocumentId(workFlowDocumentRef.getDocumentId());
        workflowMessageCO.setApprovalText(workFlowDocumentRef.getRejectReason());
        workflowMessageCO.setRemark("单据编号:" + workFlowDocumentRef.getDocumentNumber());
        workflowMessageCO.setDocumentTypeId(workFlowDocumentRef.getDocumentTypeId());
        workflowMessageCO.setDocumentTypeCode(workFlowDocumentRef.getDocumentTypeCode());
 
        String originalSevice = applicationName + ":**";
        String destinationService = workFlowDocumentRef.getDestinationService();
        WorkflowCustomRemoteEvent workflowCustomRemoteEvent = new WorkflowCustomRemoteEvent(
                this, originalSevice, destinationService, workflowMessageCO);
        return workflowCustomRemoteEvent;
    }

由 WorkflowCustomRemoteEvent 实体来封装要发送的消息,该类实现了 ApplicationEvent类。


//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//
 
package com.hand.hcf.app.mdata.client.workflow.event;
 
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import com.hand.hcf.app.mdata.client.workflow.dto.WorkflowMessageCO;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
 
@JsonTypeInfo(
    use = Id.NAME,
    property = "type"
)
@JsonIgnoreProperties({"source"})
public class WorkflowCustomRemoteEvent extends RemoteApplicationEvent {
    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss:SSS");
    private WorkflowMessageCO workflowMessage;
 
    public WorkflowCustomRemoteEvent(Object source, String originService, String destinationService, WorkflowMessageCO workflowMessage) {
        super(source, originService, destinationService);
        this.workflowMessage = workflowMessage;
    }
 
    public String toString() {
        return "WorkflowCustomRemoteEvent{WorkflowMessageCO=" + this.workflowMessage + ",eventId:" + super.getId() + ",originService:" + super.getOriginService() + ",destinationService:" + super.getDestinationService() + ",time:" + simpleDateFormat.format(new Date(super.getTimestamp())) + '}';
    }
 
    public WorkflowMessageCO getWorkflowMessage() {
        return this.workflowMessage;
    }
 
    public void setWorkflowMessage(final WorkflowMessageCO workflowMessage) {
        this.workflowMessage = workflowMessage;
    }
 
    public boolean equals(final Object o) {
        if (o == this) {
            return true;
        } else if (!(o instanceof WorkflowCustomRemoteEvent)) {
            return false;
        } else {
            WorkflowCustomRemoteEvent other = (WorkflowCustomRemoteEvent)o;
            if (!other.canEqual(this)) {
                return false;
            } else {
                Object this$workflowMessage = this.getWorkflowMessage();
                Object other$workflowMessage = other.getWorkflowMessage();
                if (this$workflowMessage == null) {
                    if (other$workflowMessage != null) {
                        return false;
                    }
                } else if (!this$workflowMessage.equals(other$workflowMessage)) {
                    return false;
                }
 
                return true;
            }
        }
    }
 
    protected boolean canEqual(final Object other) {
        return other instanceof WorkflowCustomRemoteEvent;
    }
 
    public int hashCode() {
        int PRIME = true;
        int result = 1;
        Object $workflowMessage = this.getWorkflowMessage();
        int result = result * 59 + ($workflowMessage == null ? 43 : $workflowMessage.hashCode());
        return result;
    }
 
    public WorkflowCustomRemoteEvent() {
    }
}

代码解析:

WorkflowCustomRemoteEvent 继承了 RemoteApplicationEvent 类，RemoteApplicationEvent 该类是 spring cloud bus包下的一个接收发送消息的事件类,同时该类继承了spring -context的ApplicationEvent 类，引入spring cloud bus包，就可以使用RemoteApplicationEvent 向绑定的rabbitmq队列发送消息。

spring cloud的bus依赖:


<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

Spring Cloud Bus包里的RemoteApplicationEvent类,该类用来监听远程的事件:

@JsonIgnoreProperties 注解表示属性值source不会被解析成json数据:


//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//
 
package org.springframework.cloud.bus.event;
 
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import java.util.UUID;
import org.springframework.context.ApplicationEvent;
import org.springframework.util.StringUtils;
 
@JsonTypeInfo(
    use = Id.NAME,
    property = "type"
)
@JsonIgnoreProperties({"source"})
public abstract class RemoteApplicationEvent extends ApplicationEvent {
    private static final Object TRANSIENT_SOURCE = new Object();
    private final String originService;
    private final String destinationService;
    private final String id;
 
    protected RemoteApplicationEvent() {
        this(TRANSIENT_SOURCE, (String)null, (String)null);
    }
 
    protected RemoteApplicationEvent(Object source, String originService, String destinationService) {
        super(source);
        this.originService = originService;
        if (destinationService == null) {
            destinationService = "**";
        }
 
        if (!"**".equals(destinationService) && StringUtils.countOccurrencesOf(destinationService, ":") <= 1 && !StringUtils.endsWithIgnoreCase(destinationService, ":**")) {
            destinationService = destinationService + ":**";
        }
 
        this.destinationService = destinationService;
        this.id = UUID.randomUUID().toString();
    }
 
    protected RemoteApplicationEvent(Object source, String originService) {
        this(source, originService, (String)null);
    }
 
    public String getOriginService() {
        return this.originService;
    }
 
    public String getDestinationService() {
        return this.destinationService;
    }
 
    public String getId() {
        return this.id;
    }
 
    public int hashCode() {
        int prime = true;
        int result = 1;
        int result = 31 * result + (this.destinationService == null ? 0 : this.destinationService.hashCode());
        result = 31 * result + (this.id == null ? 0 : this.id.hashCode());
        result = 31 * result + (this.originService == null ? 0 : this.originService.hashCode());
        return result;
    }
 
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        } else if (obj == null) {
            return false;
        } else if (this.getClass() != obj.getClass()) {
            return false;
        } else {
            RemoteApplicationEvent other = (RemoteApplicationEvent)obj;
            if (this.destinationService == null) {
                if (other.destinationService != null) {
                    return false;
                }
            } else if (!this.destinationService.equals(other.destinationService)) {
                return false;
            }
 
            if (this.id == null) {
                if (other.id != null) {
                    return false;
                }
            } else if (!this.id.equals(other.id)) {
                return false;
            }
 
            if (this.originService == null) {
                if (other.originService != null) {
                    return false;
                }
            } else if (!this.originService.equals(other.originService)) {
                return false;
            }
 
            return true;
        }
    }
}

业务逻辑收到消息源码分析:

使用@EventListener注解来监听事件，如果收到事件，会在WorkflowCustomRemoteEvent类里进行回调，就到指定对应WorkflowEventConsumerInterface的实现类里进行处理:


//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//
 
package com.hand.hcf.app.mdata.client.workflow.event;
 
import org.springframework.context.event.EventListener;
 
public interface WorkflowEventConsumerInterface {
    @EventListener({WorkflowCustomRemoteEvent.class})
    void workFlowConsumer(WorkflowCustomRemoteEvent event);
}

WorkflowCustomRemoteEvent完整代码:


 //
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//
 
package com.hand.hcf.app.mdata.client.workflow.event;
 
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
import com.hand.hcf.app.mdata.client.workflow.dto.WorkflowMessageCO;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
 
@JsonTypeInfo(
    use = Id.NAME,
    property = "type"
)
@JsonIgnoreProperties({"source"})
public class WorkflowCustomRemoteEvent extends RemoteApplicationEvent {
    private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss:SSS");
    private WorkflowMessageCO workflowMessage;
 
    public WorkflowCustomRemoteEvent(Object source, String originService, String destinationService, WorkflowMessageCO workflowMessage) {
        super(source, originService, destinationService);
        this.workflowMessage = workflowMessage;
    }
 
    public String toString() {
        return "WorkflowCustomRemoteEvent{WorkflowMessageCO=" + this.workflowMessage + ",eventId:" + super.getId() + ",originService:" + super.getOriginService() + ",destinationService:" + super.getDestinationService() + ",time:" + simpleDateFormat.format(new Date(super.getTimestamp())) + '}';
    }
 
    public WorkflowMessageCO getWorkflowMessage() {
        return this.workflowMessage;
    }
 
    public void setWorkflowMessage(final WorkflowMessageCO workflowMessage) {
        this.workflowMessage = workflowMessage;
    }
 
    public boolean equals(final Object o) {
        if (o == this) {
            return true;
        } else if (!(o instanceof WorkflowCustomRemoteEvent)) {
            return false;
        } else {
            WorkflowCustomRemoteEvent other = (WorkflowCustomRemoteEvent)o;
            if (!other.canEqual(this)) {
                return false;
            } else {
                Object this$workflowMessage = this.getWorkflowMessage();
                Object other$workflowMessage = other.getWorkflowMessage();
                if (this$workflowMessage == null) {
                    if (other$workflowMessage != null) {
                        return false;
                    }
                } else if (!this$workflowMessage.equals(other$workflowMessage)) {
                    return false;
                }
 
                return true;
            }
        }
    }
 
    protected boolean canEqual(final Object other) {
        return other instanceof WorkflowCustomRemoteEvent;
    }
 
    public int hashCode() {
        int PRIME = true;
        int result = 1;
        Object $workflowMessage = this.getWorkflowMessage();
        int result = result * 59 + ($workflowMessage == null ? 43 : $workflowMessage.hashCode());
        return result;
    }
 
    public WorkflowCustomRemoteEvent() {
    }
}

准备接收消息:


//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//
 
package com.hand.hcf.app.mdata.client.workflow.event;
 
import com.hand.hcf.app.mdata.base.util.OrgInformationUtil;
import com.hand.hcf.app.mdata.client.workflow.dto.ApprovalNotificationCO;
import com.hand.hcf.app.mdata.client.workflow.dto.ApprovalResultCO;
import com.hand.hcf.app.mdata.client.workflow.dto.WorkflowMessageCO;
import com.hand.hcf.app.mdata.client.workflow.enums.DocumentOperationEnum;
import com.hand.hcf.core.exception.BizException;
import com.hand.hcf.core.security.domain.PrincipalLite;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
 
public abstract class AbstractWorkflowEventConsumerInterface implements WorkflowEventConsumerInterface {
    @Value("${spring.application.name:}")
    private String applicationName;
    private static final Logger logger = LoggerFactory.getLogger(AbstractWorkflowEventConsumerInterface.class);
 
    public AbstractWorkflowEventConsumerInterface() {
    }
 
    @Transactional(
        rollbackFor = {Exception.class}
    )
    public void workFlowConsumer(WorkflowCustomRemoteEvent workflowCustomRemoteEvent) {
        String destinationService = this.applicationName + ":**";
        WorkflowMessageCO workflowMessage = workflowCustomRemoteEvent.getWorkflowMessage();
        if (destinationService.equalsIgnoreCase(workflowCustomRemoteEvent.getDestinationService()) && workflowMessage.getStatus() > DocumentOperationEnum.APPROVAL.getId()) {
            if (logger.isInfoEnabled()) {
                logger.info("接收到工作流事件消息：" + workflowCustomRemoteEvent);
            }
 
            PrincipalLite userBean = workflowMessage.getUserBean();
            OrgInformationUtil.setAuthentication(userBean);
            this.doWorkFlowConsumer(workflowCustomRemoteEvent, workflowMessage);
        }
 
    }
 
    protected void doWorkFlowConsumer(WorkflowCustomRemoteEvent workflowCustomRemoteEvent, WorkflowMessageCO workflowMessage) {
        ApprovalNotificationCO approvalNotificationCO = new ApprovalNotificationCO();
        approvalNotificationCO.setDocumentId(workflowMessage.getDocumentId());
        approvalNotificationCO.setDocumentOid(workflowMessage.getEntityOid());
        approvalNotificationCO.setDocumentCategory(Integer.parseInt(workflowMessage.getEntityType()));
        approvalNotificationCO.setDocumentStatus(workflowMessage.getStatus());
        approvalNotificationCO.setDocumentTypeId(workflowMessage.getDocumentTypeId());
        approvalNotificationCO.setDocumentTypeCode(workflowMessage.getDocumentTypeCode());
        ApprovalResultCO approvalResultCO = this.approve(approvalNotificationCO);
        if (Boolean.FALSE.equals(approvalResultCO.getSuccess())) {
            throw new BizException(approvalResultCO.getError());
        }
    }
 
    @PostMapping({"/api/implement/workflow/approve"})
    public abstract ApprovalResultCO approve(@RequestBody ApprovalNotificationCO approvalNoticeCO);
}

最后在实现 AbstractWorkflowEventConsumerInterface 的类中,重写 approve 方法:


package com.hand.hcf.app.expense.common.workflow;
 
import com.codingapi.txlcn.tc.annotation.LcnTransaction;
import com.hand.hcf.app.apply.prepayment.dto.CashPaymentRequisitionHeaderCO;
import com.hand.hcf.app.client.org.OrganizationInterface;
import com.hand.hcf.app.client.user.UserClient;
import com.hand.hcf.app.expense.accrual.service.ExpenseAccrualHeaderService;
import com.hand.hcf.app.expense.adjust.service.ExpenseAdjustHeaderService;
import com.hand.hcf.app.expense.application.service.ApplicationHeaderService;
import com.hand.hcf.app.expense.client.AccountingClient;
import com.hand.hcf.app.expense.client.extraApi.FecPeripheralInterface;
import com.hand.hcf.app.expense.client.extraApi.GetWorkflowInterface;
import com.hand.hcf.app.expense.common.domain.enums.DocumentTypeEnum;
import com.hand.hcf.app.expense.common.externalApi.OrganizationService;
import com.hand.hcf.app.expense.common.externalApi.PrepaymentService;
import com.hand.hcf.app.expense.common.utils.SyncLockPrefix;
import com.hand.hcf.app.expense.report.domain.ExpenseReportHeader;
import com.hand.hcf.app.expense.report.service.ExpenseReportHeaderService;
import com.hand.hcf.app.expense.report.service.ExpenseReportPrintInfoService;
import com.hand.hcf.app.expense.sftp.SFTP;
import com.hand.hcf.app.expense.sftp.SftpConfig;
import com.hand.hcf.app.expense.travel.service.TravelApplicationHeaderService;
import com.hand.hcf.app.mdata.base.util.OrgInformationUtil;
import com.hand.hcf.app.mdata.client.contact.ContactClient;
import com.hand.hcf.app.mdata.client.workflow.WorkflowClient;
import com.hand.hcf.app.mdata.client.workflow.WorkflowInterface;
import com.hand.hcf.app.mdata.client.workflow.dto.ApprovalErrorDataCO;
import com.hand.hcf.app.mdata.client.workflow.dto.ApprovalNotificationCO;
import com.hand.hcf.app.mdata.client.workflow.dto.ApprovalResultCO;
import com.hand.hcf.app.mdata.client.workflow.dto.WorkFlowDocumentRefCO;
import com.hand.hcf.app.mdata.client.workflow.enums.DocumentOperationEnum;
import com.hand.hcf.app.mdata.client.workflow.event.AbstractWorkflowEventConsumerInterface;
import com.hand.hcf.core.exception.BizException;
import com.hand.hcf.core.redisLock.annotations.LockedObject;
import com.hand.hcf.core.redisLock.annotations.SyncLock;
import com.jcraft.jsch.ChannelSftp;
import jline.internal.Log;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
 
import java.io.*;
import java.time.ZonedDateTime;
import java.util.Vector;
 
 
@RestController
public class WorkflowEventConsumer extends AbstractWorkflowEventConsumerInterface {
 
 // 接收消息队列的消息
    @LcnTransaction
    @Transactional(rollbackFor = Exception.class)
    @SyncLock(lockPrefix = SyncLockPrefix.PUBLIC_REPORT)
    @Override
    public ApprovalResultCO approve(@LockedObject("documentId")@RequestBody ApprovalNotificationCO approvalNoticeCO) { }
 
}

最后在approver方法里执行业务逻辑。

四、死信队列

1.什么是死信队列？

死信队列也是一种队列，用来存放未经消费的消息，可以设置一段时间后，会重新路由到其他设置的队列中,可以运用死信机制来实现延迟队列。

2.哪些情况消息会进入到死信队列中?

生产者产生的消息被拒绝消费后，会进入到死信队列中。
通过TTL设置的key超过最大的有效时间后，会进入到死信队列中。


channel.QueueDeclare(queue: "q1",
                                     durable: false,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: new Dictionary<string, object> {                                       
                                        { "x-message-ttl",10000} 
           //x-message-ttl即设置当前队列消息的过期时间。
                                     });

队列中的消息存放满后，消息会进入到死信队列中。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/438764