RabbitMQ

我在windows平臺下安裝

整合RabbitMQ

要先安裝 erlang，要依賴他

啟動：

安裝了管理介面後
rabbitmq-plugins enable rabbitmq_management

SpringBoot整合 RabbitMQ

1.新增依賴 spring-boot-starter-amqp

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

新增配置

#rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
#消費者數量
spring.rabbitmq.listener.simple.concurrency= 10
spring.rabbitmq.listener.simple.max-concurrency= 10
#消費者每次從佇列獲取的訊息數量
 

spring.rabbitmq.listener.simple.prefetch= 1
#消費者自動啟動
spring.rabbitmq.listener.simple.auto-startup=true
#消費失敗，自動重新入隊
spring.rabbitmq.listener.simple.default-requeue-rejected= true
#啟用傳送重試
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=1000
spring.rabbitmq.template
 
.retry.max-attempts=3
spring.rabbitmq.template.retry.max-interval=10000
spring.rabbitmq.template.retry.multiplier=1.0

配置類

@Configuration
public class MQConfig {
    public static final String QUEUE ="queue";

    //佇列
    @Bean
    public Queue queue(){
        return new Queue("queue",true);
    }

2.建立訊息接受者

@Service
public class MQReceiver {

    private static Logger log = LoggerFactory.getLogger(MQReceiver.class);
        @RabbitListener(queues = MQConfig.QUEUE)
    public void receiver(String message){
        log.info("recevier"+message);
    }
}

3.建立訊息傳送者

public class MQSender {
    @Autowired
    AmqpTemplate amqpTemplate;

    @Autowired
    RedisService redisService;
      private static Logger log = LoggerFactory.getLogger(MQReceiver.class);

    public void send(Object message){
        //將bean轉為字串
        String msg = redisService.beanToString(message);
        //傳送
        amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);
        log.info("send:"+msg);

    }
}

使用 guest預設是不允許遠端連線的。

RabbitMQ4種交換機模式

訊息是先發到交換機上，再由交換機發送到佇列

SirectExchange 安裝routingkey分發到指定佇列

Direct模式

配置

//佇列
@Bean
public Queue queue(){
    return new Queue(QUEUE,true);
}

傳送

public void send(Object message){
    //將bean轉為字串
    String msg = redisService.beanToString(message);
    //傳送
    amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);
    log.info("send:"+msg);

}

接收

@RabbitListener(queues = MQConfig.QUEUE)
public void receiver(String message){
    log.info("recevier"+message);
}

Topic模式

多關鍵字匹配
配置：

public static final String TOPIC_QUEUE1 ="topic.queue1";public static final String TOPIC_QUEUE2 ="topic.queue2";public static final String TOPIC_EXCHANGE="topicExchange";public static final String ROUTING_KEY1="topic.key1";public static final String ROUTING_KEY2="topic.#";//# 萬用字元//Topic模式 交換機 [email protected] Queue topicQueue1(){    return new Queue(TOPIC_QUEUE1,true);}@Beanpublic Queue topicQueue2(){    return new Queue(TOPIC_QUEUE2,true);}//topic [email protected] TopicExchange topicExchange(){    return new TopicExchange(TOPIC_EXCHANGE);}//繫結@Beanpublic Binding topicBinding1(){    return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(ROUTING_KEY1);}@Beanpublic Binding topicBinding2(){    return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(ROUTING_KEY2);}


傳送



public void sendTpoic(Object message){
    //將bean轉為字串
    String msg = redisService.beanToString(message);
    //傳送
    amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,MQConfig.ROUTING_KEY1,msg+":1");
    amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,MQConfig.ROUTING_KEY2,msg+":2");
    log.info("send topic:"+msg);

}

接收

@RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
public void receiverTopic1(String message){
    log.info("recevier topic 1:"+message);
}

@RabbitListener(queues = MQConfig.TOPIC_QUEUE2)
public void receiverTopic2(String message){
    log.info("recevier topic 2:"+message);
}

Fanout 廣播模式
將訊息分發到所有的繫結佇列，無routingkey的概念
還是繫結的之前兩個主題模式建立的佇列

配置

//Fanout模式 交換機Exchange（廣播）
@Bean
public FanoutExchange fanoutExchange(){
    return new FanoutExchange(FANOUT_EXCHANGE);
}

//繫結
@Bean
public Binding fanoutBinding1(){
    return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding2(){
    return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());
}

傳送

public void sendFanout(Object message){
    //將bean轉為字串
    String msg = redisService.beanToString(message);
    //傳送
    amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg);
    log.info("send fanout: :"+msg);

}

接收
還是對那兩個主題的佇列進行監聽

@RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
public void receiverTopic1(String message){
    log.info("recevier topic 1:"+message);
}

@RabbitListener(queues = MQConfig.TOPIC_QUEUE2)
public void receiverTopic2(String message){
    log.info("recevier topic 2:"+message);
}

Headers

通過新增屬性 key-value匹配

配置

//Hander模式


@Bean
public HeadersExchange headerstExchange(){
    return new HeadersExchange(HEADERS_EXCHANGE);
}

@Bean
public Queue headerqueue(){
    return new Queue(HEADERS_QUEUE,true);
}


@Bean
public Binding headersBinding(){
    Map<String,Object> map  = new HashMap<String, Object>();
    map.put("header1","value1");
    map.put("header2","value2");
    //滿足key-value才會往佇列中放東西
    return BindingBuilder.bind(headerqueue()).to(headerstExchange()).whereAll(map).match();
}

傳送

public void sendHeader(Object message){
    //將bean轉為字串
    String msg = redisService.beanToString(message);
    //傳送
    MessageProperties properties = new MessageProperties();
    properties.setHeader("header1","value1");
    properties.setHeader("header2","value2");
    Message obj = new Message(msg.getBytes(),properties);
    amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE,"",obj);
    log.info("send header: :"+msg);

}

接收

@RabbitListener(queues = MQConfig.HEADERS_QUEUE)
public void receiverHeader(byte[] message){
    log.info("recevier header :"+new String (message));
}

使用RabbitMQ改寫秒殺功能

思路：減少資料庫訪問

1.系統初始化，把商品庫存數量載入到Redis

2.收到請求，Redis預減庫存，庫存不足，直接返回，否則3

3.請求入隊，立即返回排隊中

4.請求出隊，生成訂單，減少庫存

5.客戶端輪詢，是否秒殺成功

系統初始化，把商品庫存數量載入到Redis

如何在初始化的時候就將庫存資料存入快取中

實現這個藉口 InitializingBean

public class MiaoshaController implements InitializingBean{

然後
實現方法

//在系統初始化的時候呼叫
@Override
public void afterPropertiesSet() throws Exception {
    List<GoodsVo> goodsVoList =  goodsService.listGoodsVo();
    if (goodsVoList==null){
        return ;
         }
    for (GoodsVo goodsVo : goodsVoList){
        //將商品庫存載入到redis中  
          redisService.set(GoodsKey.getMiaoshaGoodsStock,""+goodsVo.getId(),goodsVo.getStockCount());
    }
}      

收到請求，Redis預減庫存，庫存不足，直接返回，否則3 請求入隊，立即返回排隊中

//減少庫存
long stock = redisService.decr(GoodsKey.getMiaoshaGoodsStock,""+goodsId);
if (stock<0){//如果沒有庫存了就返回秒殺失敗了。
    return Result.error(CodeMsg.MIAO_SHA_OVER);
}
//判斷是否秒殺到了。
MiaoshaOrder order  = orderService.getMiaoshaOrderByUserIdGoodsId(user.getId(),goodsId);
    if(order!=null) {//已經秒殺到了
        return Result.error(CodeMsg.REPEATE_MIAOSHA);
    }
//入隊
MiaoshaMessage mm = new MiaoshaMessage();
mm.setGoodsId(goodsId);
mm.setUser(user);
mqSender.sendMiaoshaMessage(mm);
return  Result.success(0);//排隊中
}

請求出隊，生成訂單，減少庫存

接收

   @RabbitListener(queues = MQConfig.MIAOSHA_QUEUE)
    public void receiver(String message) {
        log.info("recevier" + message);
        //將訊息還原為bean物件
        MiaoshaMessage mm = redisService.stringToBean(message, MiaoshaMessage.class);
        MiaoshaUser user = mm.getUser();
        long goodsId = mm.getGoodsId();
        //減庫存，訂單
        GoodsVo goods = goodsService.getGoodsVoByGoodsId(goodsId);
        int stock = goods.getGoodsStock();
        if (stock <= 0) {//沒有庫存了.
            return;
        }
              //判斷是否重複秒殺
        MiaoshaOrder order = orderService.getMiaoshaOrderByUserIdGoodsId(user.getId(), goodsId);
        if (order != null) {
            return;
        } else {
            //秒殺
            //減庫存 下訂單 寫入秒殺訂單
             miaoshaService.miaosha(user, goods);
        }

    }  

客戶端輪詢，是否秒殺成功

定義個介面

//秒殺排隊後輪詢
    //秒殺成功返回訂單id
    //秒殺失敗返回 -1
    //排隊中 返回 0
    @RequestMapping("/result")
    public Result miaoshResult(Model model, MiaoshaUser user, @RequestParam("goodsId") long goodsId) {
        //沒登入就跳轉到登入頁面
        model.addAttribute("user", user);
        if (user == null) {
            return Result.error(CodeMsg.SESSION_ERROR);
        }
//查詢下是否生成了訂單
        long rsult = miaoshaService.getMiaoshResult(user.getId(), goodsId);
        return Result.success(rsult);


    }

miaoshaService.getMiaoshResult

public long getMiaoshResult(Long userId, long goodsId) {
    MiaoshaOrder order = orderService.getMiaoshaOrderByUserIdGoodsId(userId,goodsId);
    if (order!=null){
        return order.getId();
    }else {
        boolean isOver = getGoodsOver(goodsId);
        if (isOver){//賣完了還在沒有訂單
            return -1;
        }else {
            return 0;
        }

    }

沒有訂單有兩種情況，賣完了失敗，和排隊中

在上面的秒殺那做個標記。這個商品是否秒殺完了。存入redis中。
之後去判斷是否存在這個key就知道是哪種情況

@Transactional
public OrderInfo miaosha(MiaoshaUser user, GoodsVo goods) {
    //減少庫存 下訂單 寫入秒殺訂單
    boolean success = goodsServic.reduceStock(goods);
    //減庫存成功了才進行下訂單
    if (success) {
        //訂單表，秒殺表
        OrderInfo orderInfo = orderService.createOrder(user, goods);
        return orderInfo;
    } else {//說明商品秒殺完了。做一個標記
        setGoodsOver(goods.getId());
        return null;
    }

}

public void setGoodsOver(Long goodsId) {
    redisService.set(MiaoshaKey.isGoodsOver,""+goodsId,true);


}

private boolean getGoodsOver(long goodsId) {
    return redisService.exists(MiaoshaKey.isGoodsOver,""+goodsId);
}

前端請求做判斷。