@RestController
@RequestMapping
public class RedisController {


    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private SysUserMapper sysUserMapper;


    @GetMapping
    public void duobleCancle() throws InterruptedException {

        //刪除緩存
        redisTemplate.delete("1");

        //獲取數(shù)據(jù)
        SysUser sysUser = sysUserMapper.selectUserById(Long.valueOf(1));

        //更新數(shù)據(jù)
        SysUser updateSysUser =new SysUser();
        updateSysUser.setUserName("Lxlxxx");
        updateSysUser.setEmail("@163.com");

        UpdateWrapper<SysUser> updateWrapper = new UpdateWrapper();
        updateWrapper.eq("userId",1);

        sysUserMapper.update(updateSysUser,updateWrapper);

        //休眠時(shí)間這里的設(shè)定，是根據(jù)讀業(yè)務(wù)的同步時(shí)間來設(shè)定的，這是一個(gè)大概的范圍
        Thread.sleep(3000);

        //將讀取的數(shù)據(jù)放入緩存
        redisTemplate.opsForValue().append(sysUser.getUserId(), JSON.toJSONString(sysUser));


        //第二次刪除緩存
        //如果刪除失敗，緩存里面存的應(yīng)該還是舊數(shù)據(jù)，后面線程讀的還是舊數(shù)據(jù)，實(shí)際數(shù)據(jù)庫已經(jīng)是新數(shù)據(jù)了
        redisTemplate.delete("1");
    }

#編輯conf/canal.properties，修改MQ配置
canal.ip = 1 #canal服務(wù)器標(biāo)識
canal.serverMode = rabbitmq # 指定rabbitmq
canal.mq.servers = 127.0.0.1 ##
canal.mq.vhost=canal  #MQ虛擬機(jī)名稱
canal.mq.exchange=exchange.trade #交換機(jī)名稱，用于將消息發(fā)送到綁定的隊(duì)列
canal.mq.username=guest #MQ登錄賬號，注意要有上面vhost的權(quán)限
canal.mq.password=guest #MQ密碼
---------------------------------------------------------------------------------
#編輯conf/example/instance.properties實(shí)例配置,配置數(shù)據(jù)庫信息
canal.instance.dbUsername=root
canal.instance.dbPassword=123456
canal.instance.mysql.slaveId=1234 
canal.instance.master.address=127.0.0.1:3306 ## 數(shù)據(jù)庫地址
canal.instance.defaultDatabaseName=test ## 數(shù)據(jù)庫名
canal.mq.topic=example # 路由鍵，需要跟MQ中交換機(jī)隊(duì)列的綁定路由key保持一致

#增加以下配置
log-bin=mysql-bin # 開啟 binlog
binlog-format=ROW # 選擇 ROW 模式
server_id=1 # 配置 

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
?
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
?
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
?
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.3</version>
        </dependency>
                <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>${mybatis.plus.version}</version>
        </dependency>

spring:
  rabbitmq:
    virtual-host: canal
    host: 127.0.0.1
    publisher-confirms: true
  #數(shù)據(jù)源
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=UTC
    username: root
    password: root
    driver-class-name: com.mysql.jdbc.Driver
  redis:
    host: 127.0.0.1
?

@Configuration
public class RabbitMqConfig {
    @Bean
    public Queue TestDirectQueue() {
        return new Queue("exchange.canal.queue",true);
    }
    //Direct交換機(jī) 起名：exchange.trade
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("exchange.canal");
    }
    //綁定  隊(duì)列和交換機(jī)綁定
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("example");
    }
}

@Component
@Slf4j
public class RabbitMqListener {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @RabbitListener(queues = "exchange.canal.queue")
    public void process(Message message) {
        log.info("canal queue消費(fèi)的消息" + message.getBody());
        Map map = JSON.parseObject(message.getBody(), Map.class);
        JSONArray array = null;
        String sqlType = (String) map.get("type");
        //如果是查詢,則去獲取data數(shù)據(jù)
        if (StringUtils.endsWithIgnoreCase("SELECT", sqlType)) {
            array = JSONArray.parseArray((String) map.get("data"));
        }
        if (null == array) {
            return;
        }
        JSONObject jsonObject = array.getJSONObject(0);
        //如果是update、insert 其中一個(gè)則去更新緩存
        if (StringUtils.endsWithIgnoreCase("UPDATE", sqlType)
                || StringUtils.endsWithIgnoreCase("INSERT", sqlType)) {
            redisTemplate.boundValueOps(jsonObject.get("code").toString()).set(jsonObject.toString());
            //反之類型為"delete"則去刪除緩存
        } else if (StringUtils.endsWithIgnoreCase("DELETE", sqlType)) {
            redisTemplate.delete(jsonObject.get("code").toString());
        }
        //查詢到新的數(shù)據(jù)去更新緩存 ，反之再去刪除對應(yīng)的緩存，這里進(jìn)行二次刪除
        if (StringUtils.endsWithIgnoreCase("SELECT", sqlType)) {
            redisTemplate.boundValueOps(jsonObject.get("code").toString()).set(jsonObject.toString());
        } else {
            redisTemplate.delete(jsonObject.get("code").toString());
        }
    }
}