Springboot项目整合Ribbitmq详细教程

Posted 2021-11-01 洛阳泰山

首先在springboot的pom文件里引入依赖

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

然后application.yml：

ps：里面的虚拟host配置项不是必须的，我自己在rabbitmq服务上创建了自己的虚拟host，所以我配置了；你们不创建，就不用加这个配置项。

spring:
  application:
    name: 3d-gis
  datasource:
    driver-class-name: org.postgresql.Driver
    url: jdbc:postgresql://172.16.10.201:5432/3d_gis
    username: postgres
    password: postgres
  rabbitmq:
    host: 172.16.10.201
    port: 5672
    username: leaniot
    password: leaniot
    virtual-host: /3d_gis
    listener:
      simple:
        acknowledge-mode: manual
      direct:
        acknowledge-mode: manual

RabbitMQ配置类

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * RabbitMQ 配置
 *
 * @author tarzan
 * @version 1.0
 * @date 2021年05月18日 09:15:40
 * @since JDK1.8
 */

@Configuration
public class RabbitMQConfig implements InitializingBean {

    /** gis图形交换机 */
    public static final String  GIS_GRAPHICS_EXCHANGE = "3d_gis_exchange";
    /** gis图形数据接收序列 */
    public static final String GIS_DATA_RECEIVE_REQUEUE = "gis_data_receive_queue";
    /** gis图形数据返回（发送）序列 */
    public static final String GIS_DATA_SEND_QUEUE = "gis_data_send_queue";

    @Bean
    public Queue gisDataReceiveQueue () {
        return new Queue(GIS_DATA_RECEIVE_REQUEUE);
    }

    @Bean
    public Queue gisDataSendQueue () {
        return new Queue(GIS_DATA_SEND_QUEUE);
    }


    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(GIS_GRAPHICS_EXCHANGE);
    }

    @Bean
    public Binding receiveBinding () {
        return BindingBuilder.bind(gisDataReceiveQueue()).to(directExchange()).withQueueName();
    }

    @Bean
    public Binding sendBinding () {
        return BindingBuilder.bind(gisDataSendQueue()).to(directExchange()).withQueueName();
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }
}

RabbitSender类

import java.util.UUID;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class RabbitSender {

    @Resource
    private RabbitTemplate rabbitTemplate;

    /**
     * 	这里就是确认消息的回调监听接口，用于确认消息是否被broker所收到
     */
    final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        /**
         * 	@param ack broker 是否落盘成功
         * 	@param cause 失败的一些异常信息
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.err.println("消息ACK结果:" + ack + ", correlationData: " + correlationData.getId());
        }
    };

    /**
     * 	对外发送消息的方法
     * @param message 	具体的消息内容
     * @throws Exception
     */
    public void send(Object message)  {
        CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(RabbitMQConfig.GIS_GRAPHICS_EXCHANGE, RabbitMQConfig.GIS_DATA_RECEIVE_REQUEUE, message, data);
    }
}

RabbitReceiver类

import org.springblade.core.tool.utils.Charsets;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.core.Message;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
import java.io.IOException;

@Component
public class RabbitReceiver {

    /**
     * 	组合使用监听
     * @param message
     * @param channel
     */
    @RabbitListener(queues =RabbitMQConfig.GIS_DATA_RECEIVE_REQUEUE)
    public void onMessage(Message message, Channel channel) {
        //	1. 收到消息以后进行业务端消费处理
        System.err.println("-----------------------");
        System.err.println("消费消息:" +  new String(message.getBody(), Charsets.UTF_8));

        //  2. 处理成功之后 获取deliveryTag 并进行手工的ACK操作, 因为我们配置文件里配置的是 手工签收
        Long deliveryTag = message.getMessageProperties().getDeliveryTag();
       try {
           channel.basicAck(deliveryTag, false);
       } catch (IOException e) {
           new IOException();
       }
   }

}

测试类

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.AllArgsConstructor;
import org.springblade.core.secure.annotation.NoToken;
import org.springblade.core.tool.api.R;
import org.springblade.three.dimension.gis.rabbitmq.RabbitSender;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@NoToken
@RestController
@AllArgsConstructor
@RequestMapping("rabbit")
@Api(value = "消息队列", tags = "消息队列")
public class RabbitController {
    @Resource
    private RabbitSender sender;


    @ApiOperation(value = "添加 @author Tarzan Liu")
    @GetMapping("test")
    public R test(){
        sender.send("测试666");
        return R.status(true);
    }
}

测试结果