RabbitMQ实现削峰限流demo

迈不过友情╰ 2023-07-17 15:44 7阅读 0赞

## ***削峰限流实现方式：固定队列长度，固定数据数量*** ##

## 一、controller类中模拟10000个线程 ##

package com.woniuxy.handler;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.woniuxy.sender.LimitSender;
    
    @RestController
    public class CurrentLimitingHandler {
    	public static int i = 0;
    	
    	@Autowired
    	private LimitSender limitSender;
    	
    	@RequestMapping("/test")
    	public String test() {
    		//模拟10000个请求，去请求数据
    		for(i=0;i<3000;i++){
    			new Thread(new Runnable() {
    				@Override
    				public void run() {
    					//向消费者发送请求
    					limitSender.send(i);//存在多个线程发送同一个值的
    				}
    			}).start();;
    		}
    		return "";
    	}
    }

## 二、创建削峰限流交换机 ##

package com.woniuxy.configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMqConfiguration {
    
    	/*
    	 * 削峰限流
    	 */
    	@Bean
    	public Queue limitQueue() {
    		//设置队列的初始化参数
    		Map<String, Object> map = new HashMap<String, Object>();
    		//指定消息队列的长度
    		map.put("x-max-length", 100);		
    		
    		//当队列满时，多余的消息直接拒绝接收，多余的消息被丢弃
    		map.put("x-overflow", "reject-publish");	
    		
    		/*
    		 * 参数1：队列的名字
    		 * 参数2：是否为持久队列，及时rabbitmq服务停机，之后在开机该队列也存在  一般队列默认都是持久的
    		 * 参数3：是否为排他队列，是否只能由自己可见（admin），其他用户看不到该队列
    		 * 参数4：是否在不适用该队列是删除该队列
    		 * 参数5：队列初始化的参数
    		 */
    		return new Queue("limit_queue",true,false,false,map);
    		//return new Queue("limit_queue");
    	}
    	
    	@Bean
    	public FanoutExchange limitExchange() {
    		return new FanoutExchange("limit_exchange");
    	}
    	
    	@Bean
    	public Binding bindingLimitQueueToLimitExchange(Queue limitQueue,FanoutExchange limitExchange) {
    		return BindingBuilder.bind(limitQueue).to(limitExchange);
    	}
    }

## 三、limitsend类，发送方 ##

package com.woniuxy.sender;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    @Component
    public class LimitSender {
    	@Autowired
    	private AmqpTemplate amqpTemplate;
    	
    	public void send(int num) {
    		amqpTemplate.convertAndSend("limit_exchange","",num);
    	}
    }
    
    ``

## 四、接收方 ##

package com.woniuxy.receiver;
    
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class LimitReceiver { 
    	@RabbitListener(queues="limit_queue")
    	public void received(int num) { 
    		System.out.println("接收到:"+num);
    	}
    }