RabbitMQ实现项目通讯demo

小鱼儿 2023-07-17 15:42 3阅读 0赞

***以user和product为例，user请求product中数据，再返回给user***

## 一、user项目中创建交换机 ##

package com.woniuxy.configuration;

import org.springframework.amqp.core.Binding;  
import org.springframework.amqp.core.BindingBuilder;  
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 productQueue() \{  
return new Queue(“product\_queue”);  
\}  
@Bean  
//交换机  
public TopicExchange productExchange() \{  
return new TopicExchange(“product\_exchange”);  
\}  
//绑定  
@Bean  
public Binding bindingProeuctQueueToProductExchange(Queue productQueue,TopicExchange productExchange) \{  
System.out.println("======");  
return BindingBuilder.bind(productQueue).to(productExchange).with(“product.\#”);  
\}  
\}

## 二、使用Ampqtemplate,send方法，实现的是根据id查询product

public void send() {
    	String message = "1001";
    	amqpTemplate.convertAndSend("product_exchange","",message);
    }

## 三、user模块中写一个product实体类
    
    package com.woniuxy.pojo;
    
    import java.io.Serializable;
    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import lombok.experimental.Accessors;
    
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Accessors(chain=true)
    public class Product implements Serializable{
    	private static final long serialVersionUID = 1L;
    	private Integer pid;
    	private String pname;
    	private double price;
    	private Integer count;
    	private String detail;
    }
    
    ## 四、controller类，调用send方法发送查询请求

package com.woniuxy.handler;
    
    
    import java.util.List;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import org.springframework.web.client.RestTemplate;
    
    import com.woniuxy.pojo.Product;
    import com.woniuxy.sender.ProductSender;
    
    @RestController
    @RequestMapping("/user")
    public class UserHandler {

@Autowired
    private ProductSender productSender;
    
    
    @RequestMapping("/id")
    public Product findById() {
    	productSender.send();
    	return null;
    }

五、product项目中product实体

package com.woniuxy.pojo;
    
    import java.io.Serializable;
    
    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import lombok.experimental.Accessors;
    
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Accessors(chain=true)
    public class Product implements Serializable{
    	private static final long serialVersionUID = 1L;
    	private Integer pid;
    	private String pname;
    	private double price;
    	private Integer count;
    	private String detail;
    }

## 六、product项目mapper类
    
    package com.woniuxy.mapper;
    
    import java.util.List;
    
    import org.apache.ibatis.annotations.Delete;
    import org.apache.ibatis.annotations.Mapper;
    import org.apache.ibatis.annotations.Select;
    
    import com.woniuxy.pojo.Product;
    
    @Mapper		//成为mapper
    public interface ProductMapper {
    	@Select("select * from product")
    	public List<Product> all();
    	
    	//以id查询
    	@Select("select * from product where pid=#{pid}")
    	public Product findById(int pid);
    	
    	//以id删除
    	@Delete("delete from product where pid=#{pid}")
    	public int delById(int pid);
    }
    
    ## 七，product项目service类

package com.woniuxy.service;

import java.util.List;

import com.woniuxy.pojo.Product;

public interface ProductService \{  
public List all();  
public Product findById(int pid);  
public int delById(int pid);  
\}

\`\`

## 八、product项目中serviceImp类 ##

package com.woniuxy.service.impl;
    
    import java.util.List;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.cache.annotation.CacheEvict;
    import org.springframework.cache.annotation.Cacheable;
    import org.springframework.stereotype.Service;
    
    import com.woniuxy.configuration.RedisConfiguration;
    import com.woniuxy.mapper.ProductMapper;
    import com.woniuxy.pojo.Product;
    import com.woniuxy.service.ProductService;
    import com.woniuxy.utils.KeyUtil;
    import com.woniuxy.utils.RedisUtil;
    
    @Service
    public class ProductServiceImpl implements ProductService{
    	@Autowired
    	private ProductMapper productMapper;
    	@Autowired
    	private RedisUtil redisUtil;
    	
    	@Autowired
    	private KeyUtil keyUtil;
    
    	@Cacheable("allProduct")	//name的值，缓存的名字，组成key的一部分   域，命名空间
    	@Override
    	public List<Product> all() {
    		return productMapper.all();
    	}
    
    	@Cacheable("findById")
    	@Override
    	public Product findById(int pid) {
    		return productMapper.findById(pid);
    	}
    
    	@CacheEvict(value="allProduct",allEntries=true)//移除以allProduct开头所有的key
    	@Override
    	public int delById(int pid) {
    		int result =  productMapper.delById(pid);
    		if (result>0) {
    			String key = keyUtil.getKey(this, null, new Integer(pid));//方法通过反射获取
    			//调用配置类中的生成key方法生成一个key
    			//redisUtil.del(key)
    		}
    		return result;
    	}
    }
    
    ## 九。交换机

package com.woniuxy.configuration;

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 RabbitConfiguration \{  
@Bean  
public Queue userQueue() \{  
return new Queue(“user\_queue”);  
\}  
@Bean  
public FanoutExchange userTopicExchange() \{  
return new FanoutExchange(“user\_exchange”);  
\}  
//  
@Bean  
public Binding bindingUserQueueToUserTopicExchange(Queue userQueue,FanoutExchange userTopicExchange) \{  
return BindingBuilder.bind(userQueue).to(userTopicExchange);  
\}  
\}

## 十、监听，接收 ##

package com.woniuxy.receiver;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.woniuxy.pojo.Product;
    import com.woniuxy.sender.UserSender;
    import com.woniuxy.service.ProductService;
    
    @Component
    public class UserReceiver {
    	@Autowired
    	private ProductService productService;
    	
    	@Autowired
    	private UserSender userSender;
    	
    	@RabbitListener(queues="product_queue")
    	@RabbitHandler
    	public void received_id(String message) {  
    		int pid = Integer.parseInt(message);
    		System.out.println("--------2");
    		//以id查询数据
    		Product product = productService.findById(pid);
    		System.out.println(product);
    		//将该对象返回给user
    		//?
    		userSender.send(product);
    	}
    }

## 十一、product中send方法 ##

package com.woniuxy.sender;
    
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.woniuxy.pojo.Product;
    
    @Component
    public class UserSender {
    	@Autowired
    	private AmqpTemplate amqpTemplate;
    	
    	public void send(Product product) {
    		System.out.println("商品模块准备发送数据到user模块.....");
    		amqpTemplate.convertAndSend("user_exchange", "", product);
    	}
    }

## 十二，user模块接收product返回的数据· ##

package com.woniuxy.receiver;
    
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import com.woniuxy.pojo.Product;
    
    @Component
    public class ProductReceiver {
    	@RabbitListener(queues="user_queue")
    	@RabbitHandler
    	public void received(Product product) {
    		//Jackson解析
    		System.out.println("user模块接收到数据"+product);
    		
    		//利用websocket将数据推送到浏览器
    	}
    }