Springboot專案整合Ribbitmq詳細教程

首先在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);
    }
}

測驗結果