这里我用到了Redis当中的发布订阅模式实现(JAVA代码实现)

先看图示 

        

下面为代码实现

首先将RedisMessageListenerContainer交给Spring管理.

@Configuration
public class redisConfig {
    @Autowired
    RedisConnectionFactory redisConnectionFactory;

    @Autowired
    @Qualifier("carServiceDb")
    private CarService carService;//操作数据库
    @Bean
    public RedisMessageListenerContainer listenerContainer(){
        //创建一个Redis消息监听器容器对象
        RedisMessageListenerContainer listenerContainer = new RedisMessageListenerContainer();
         //设置Redis连接工厂对象
        listenerContainer.setConnectionFactory(redisConnectionFactory);

        //向监听器容器中添加一个监听器，该监听器的主题为INSERTED
        listenerContainer.addMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] pattern) {
                System.out.println("*************insert*************");

                String carJsonStr = new String(message.getBody());
                Car car = JSONObject.parseObject(carJsonStr, Car.class);

                carService.addCar(car);



                System.out.println(new String(message.getBody()));
                System.out.println(new String(pattern));
                System.out.println("*************insert*************");
            }
        }, new ChannelTopic(ChannelEnum.INSERTED.name()));

        //向监听器容器中添加一个监听器，该监听器的主题为UPDATED
        listenerContainer.addMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] pattern) {
                System.out.println("*************UPDATED*************");
                String carJsonStr = new String(message.getBody());
                Car car = JSONObject.parseObject(carJsonStr, Car.class);
                carService.updataCount(car.getBookId(),car.getNumber(),car.getUserId());

                System.out.println(new String(message.getBody()));
                System.out.println(new String(pattern));
                System.out.println("*************UPDATED*************");
            }
        }, new ChannelTopic(ChannelEnum.UPDATED.name()));

        //向监听器容器中添加一个监听器，该监听器的主题为DELETED
        listenerContainer.addMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] pattern) {
                System.out.println("*************DELETED*************");
                System.out.println(new String(message.getBody()));
                System.out.println(new String(pattern));

                List<String[]> strings = JSONArray.parseArray(new String(message.getBody()), String[].class);
                String[] bookIds = strings.get(0);
                String userId = strings.get(1)[0];
                carService.delCar(bookIds,userId);
                System.out.println("----------------------");
                System.out.println(Arrays.toString(bookIds));
                System.out.println(userId);
                System.out.println("--------delete--------");
                System.out.println("*************DELETED*************");
            }
        }, new ChannelTopic(ChannelEnum.DELETED.name()));
        return listenerContainer;
    }
}

          下面代码中的 stringRedisTemplate.convertAndSend();就是发送消息的，其中参数1是发送消息的名称，参数2是发送消息的内容

@Service("carService")
@Slf4j
public class CarServiceImpl extends ServiceImpl<CarMapper, Car> implements CarService {
    //购物车hash的Key
    public static final String CAR_KEY = "carKey";
    //hash购物车中field的前缀
    public static final String USER_CAR_HASH_FIELD_PREFIX = "car::";
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 这里存储的时候采用Redis当中的hash存储
     *
     * @param car
     */
    @Override
    @Synchronized
    public void addCar(Car car) {
        //这里存储的时候采用Redis当中的hash存储
        HashOperations<String, String, String> opsForHash = stringRedisTemplate.opsForHash();

        //这里field为CAR_HASH_FIELD
        String userCarKey = USER_CAR_HASH_FIELD_PREFIX + car.getUserId();

        //Map<String,Car>  键(存储bookId)  值存储购物车某一个商品的信息
        Map<String, Car> CarMap = null;
        boolean flag = true;
        //获得Redis中以存储的当前用户的购物车信息
        if (opsForHash.hasKey(CAR_KEY, userCarKey)) {
            //存在当前用户的购物车信息，那么获取原有的数据
            String carMapJson = opsForHash.get(CAR_KEY, userCarKey);

            CarMap = JSONObject.parseObject(carMapJson, new TypeReference<Map<String, Car>>() {
            });

            //检测当前购物车信息中是否包含新添加上的商品，如果包含则更新数量，如果不包含才新增
            if (CarMap.containsKey(car.getBookId())) {
                //获取原先商品的数量
                Integer BeforeNumber = CarMap.get(car.getBookId()).getNumber();
                log.info("BeforeNumber==============>{}", BeforeNumber);
                Integer nowNumber = car.getNumber(); //前端传过来现在的
                log.info("nowNumber==============>{}", nowNumber);
                CarMap.get(car.getBookId()).setNumber(BeforeNumber + nowNumber);

                flag = false; //存在商品
                //修改

                //包含才新增 图书的数量 在原有的基础上新增图书的数量
            } else {
                //新增
                //如果不包含当前商品信息 那么直接将商品添加到购物车信息当中
                CarMap.put(car.getBookId(), car);
            }
        } else {
            /*   新增
             *
             * 当前的用户的购物车信息不存在
             *       首先把其添加的商品首先存储到CarMap中
             * */
            CarMap = new HashMap<>();
            CarMap.put(car.getBookId(), car);
        }

        //最后将其存入redis当中
        opsForHash.put(CAR_KEY, userCarKey, JSONObject.toJSONString(CarMap));

        if (flag){
            //新增
            stringRedisTemplate.convertAndSend(ChannelEnum.INSERTED.name(), JSONObject.toJSONString(CarMap.get(car.getBookId())));
        }else {
            //修改
            stringRedisTemplate.convertAndSend(ChannelEnum.UPDATED.name(), JSONObject.toJSONString(CarMap.get(car.getBookId())));
        }
    }

    @Override
    public Collection<Car> getCarList(String userId) {
        //这里存储的时候采用Redis当中的hash存储
        HashOperations<String, String, String> opsForHash = stringRedisTemplate.opsForHash();
        String userCarKey = USER_CAR_HASH_FIELD_PREFIX + userId;
        String jsonStr = opsForHash.get(CAR_KEY, userCarKey);
        Map<String,Car> CarMap = JSONObject.parseObject(jsonStr, Map.class);
        return CarMap.values();
    }

    @Override
    public void updataCount(String bookId, int number, String userId) {
        //这里存储的时候采用Redis当中的hash存储
        HashOperations<String, String, String> opsForHash = stringRedisTemplate.opsForHash();
        String userCarKey = USER_CAR_HASH_FIELD_PREFIX + userId;

        String jsonStr = opsForHash.get(CAR_KEY, userCarKey);
        Map<String,Car> CarMap = JSONObject.parseObject(jsonStr, new TypeReference<Map<String, Car>>() {
        });
        if (!CarMap.containsKey(bookId)){ //不包含
            return;
        }

        Car car = CarMap.get(bookId);
        car.setNumber(number);

        //最后将其存入redis当中
        opsForHash.put(CAR_KEY, userCarKey, JSONObject.toJSONString(CarMap));
        stringRedisTemplate.convertAndSend(ChannelEnum.UPDATED.name(), JSONObject.toJSONString(CarMap.get(car.getBookId())));

    }

    @Override
    public void delCar(String[] bookIds, String userId) {
        //获得操作RedisHash的对象
        HashOperations<String, String, String> forHash = stringRedisTemplate.opsForHash();
        String userCarKey = USER_CAR_HASH_FIELD_PREFIX+userId;
        List<String[]> list = new ArrayList<>();
        //从redis中获取用户对应的购物车数据
        String carJsonStr = forHash.get(CAR_KEY, userCarKey);
        //修改指定商品的数量
        Map<String,Car> carMap = JSONObject.parseObject(carJsonStr, new TypeReference<Map<String,Car>>() {
        });
        for(String bookId : bookIds){
            carMap.remove(bookId);
        }

        //将修改后的数据重新添加到redis中
        forHash.put(CAR_KEY,userCarKey,JSONObject.toJSONString(carMap));
        list.add(bookIds);
        list.add(new String[]{userId});

        stringRedisTemplate.convertAndSend(ChannelEnum.DELETED.name(), JSONObject.toJSONString(list));
    }
}

上述的两段代码中

第二段代码中stringRedisTemplate.convertAndSend();就是发送消息的，其中参数1是发送消息的名称，参数2是发送消息的内容

第一段代码中的这个就是监听到后接收到的消息，其中参数1{当中的onMessage方法的参数1Message为发送消息的内容，参数2pattern是发送消息的名称}    参数2为监听指定的消息名称（这个要和stringRedisTemplate.convertAndSend()中参数1的要保持一致）

listenerContainer.addMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] pattern) {
                
            }
        }, new ChannelTopic(ChannelEnum.INSERTED.name()));