java电商项目（四）-阿里云开发者社区

[TOC]

本章内容：通过Lua、OpenResty、Nginx限流，Canal的学习和使用，实现电商门户首页的高并发解决方案

1 商城门户

1.1 搭建商城门户工程

这里我们的新职课商城前端门户项目，也采用vue和ivew研发完成，大家直接从素材导入即可，后面具体业务用到相应代码，我们在具体讲解，导入后因为现阶段没有spring cloud gateway网关微服务，我们需要修改访问后端服务的地址，直接访问真实的微服务

thymleaf,freemarker模板技术，配合网页静态化实现

门户首页商品分类展示

代码实现

1.2 门户首页分析

首页门户系统需要展示商品分类数据和广告数据，这些内容的数据需要存入redis缓存中

1560735844503

变更频率低的数据,查询频率高得数据，如何提升访问速度？

1.数据做成静态页[商品详情页]
2.做缓存[Redis]

基本的思路如下：

1560736222654

如上图此种方式简单，直接通过数据库查询数据展示给用户即可，但是通常情况下，首页（门户系统的流量一般非常的高）不适合直接通过mysql数据库直接访问的方式来获取展示。

如下思路：

1.首先访问nginx ，我们可以采用缓存的方式，先从nginx本地缓存中获取，获取到直接响应

2.如果没有获取到，再次访问redis，我们可以从redis中获取数据，如果有则返回，并缓存到nginx中

3.如果没有获取到,再次访问mysql,我们从mysql中获取数据，再将数据存储到redis中，返回。

而这里面，我们都可以使用LUA脚本嵌入到Nginx程序中执行这些查询相关的业务。

1560738068753

2 Lua(了解)

2.1 lua是什么

Lua [1] 是一个小巧的脚本语言。它是巴西里约热内卢天主教大学（Pontifical Catholic University of Rio de Janeiro）里的一个由Roberto Ierusalimschy、Waldemar Celes 和 Luiz Henrique de Figueiredo三人所组成的研究小组于1993年开发的。其设计目的是为了通过灵活嵌入应用程序中从而为应用程序提供灵活的扩展和定制功能。Lua由标准C编写而成，几乎在所有操作系统和平台上都可以编译，运行。Lua并没有提供强大的库，这是由它的定位决定的。所以Lua不适合作为开发独立应用程序的语言。Lua 有一个同时进行的JIT项目，提供在特定平台上的即时编译功能。

简单来说：

Lua 是一种轻量小巧的脚本语言，用标准C语言编写并以源代码形式开放，其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。

2.2 特性

支持面向过程(procedure-oriented)编程和函数式编程(functional programming)；
自动内存管理；只提供了一种通用类型的表（table），用它可以实现数组，哈希表，集合，对象；
语言内置模式匹配；闭包(closure)；函数也可以看做一个值；提供多线程（协同进程，并非操作系统所支持的线程）支持；
通过闭包和table可以很方便地支持面向对象编程所需要的一些关键机制，比如数据抽象，虚函数，继承和重载等。

2.3 应用场景

游戏开发
独立应用脚本
Web 应用脚本
扩展和数据库插件如：MySQL Proxy 和 MySQL WorkBench
安全系统，如入侵检测系统
redis中嵌套调用实现类似事务的功能
web容器中应用处理一些过滤缓存等等的逻辑，例如nginx。

2.4 lua的安装

有linux版本的安装也有mac版本的安装。。我们采用linux版本的安装，首先我们准备一个linux虚拟机。

安装步骤,在linux系统中执行下面的命令。

curl -R -O http://www.lua.org/ftp/lua-5.3.5.tar.gz
tar zxf lua-5.3.5.tar.gz
cd lua-5.3.5
make linux test
make install

注意：此时安装，有可能会出现如下错误：

1560739143890

此时需要安装lua相关依赖库的支持，执行如下命令即可：

yum -y install gcc gcc-c++ kernel-devel

此时再执行lua测试看lua是否安装成功

[root@localhost ~]# lua
Lua 5.1.4  Copyright (C) 1994-2008 Lua.org, PUC-Rio

2.5 入门程序

创建hello.lua文件，内容为

编辑文件hello.lua

vi hello.lua

在文件中输入：

print("hello");

保存并退出。

执行命令

lua hello.lua

输出为：

Hello

效果如下：

1564436327870

2.6 LUA的基本语法(了解)

lua有交互式编程和脚本式编程。

交互式编程就是直接输入语法，就能执行。

脚本式编程需要编写脚本，然后再执行命令执行脚本才可以。

一般采用脚本式编程。（例如：编写一个hello.lua的文件，输入文件内容，并执行lua hell.lua即可）

(1)交互式编程

Lua 提供了交互式编程模式。我们可以在命令行中输入程序并立即查看效果。

Lua 交互式编程模式可以通过命令 lua -i 或 lua 来启用：

lua -i

如下图：

1564436436450

(2)脚本式编程

我们可以将 Lua 程序代码保持到一个以 lua 结尾的文件，并执行，该模式称为脚本式编程，例如上面入门程序中将lua语法写到hello.lua文件中。

2.6.1 注释

一行注释：两个减号是单行注释:

--

多行注释：

--[[
 多行注释
 多行注释
 --]]

2.6.2 定义变量

全局变量，默认的情况下，定义一个变量都是全局变量，

如果要用局部变量需要声明为local.例如：

-- 全局变量赋值
a=1
-- 局部变量赋值
local b=2

如果变量没有初始化：则它的值为nil 这和java中的null不同。

如下图案例：

1564436763084

2.6.3 Lua中的数据类型

Lua 是动态类型语言，变量不要类型定义,只需要为变量赋值。值可以存储在变量中，作为参数传递或结果返回。

Lua 中有 8 个基本类型分别为：nil、boolean、number、string、userdata、function、thread 和 table。

数据类型	描述
nil	这个最简单，只有值nil属于该类，表示一个无效值（在条件表达式中相当于false）。
boolean	包含两个值：false和true。
number	表示双精度类型的实浮点数
string	字符串由一对双引号或单引号来表示
function	由 C 或 Lua 编写的函数
userdata	表示任意存储在变量中的C数据结构
thread	表示执行的独立线路，用于执行协同程序
table	Lua 中的表（table）其实是一个"关联数组"（associative arrays），数组的索引可以是数字、字符串或表类型。在 Lua 里，table 的创建是通过"构造表达式"来完成，最简单构造表达式是{}，用来创建一个空表。

实例：

print(type("Hello world"))      --> string
print(type(10.4*3))             --> number
print(type(print))              --> function
print(type(type))               --> function
print(type(true))               --> boolean
print(type(nil))                --> nil

2.6.4 流程控制

(1)if语句

Lua if 语句 由一个布尔表达式作为条件判断，其后紧跟其他语句组成。

语法：

if(布尔表达式)
then
   --[ 在布尔表达式为 true 时执行的语句 --]
end

实例：

1564437016055

(2)if..else语句

Lua if 语句可以与 else 语句搭配使用, 在 if 条件表达式为 false 时执行 else 语句代码块。

语法：

if(布尔表达式)
then
   --[ 布尔表达式为 true 时执行该语句块 --]
else
   --[ 布尔表达式为 false 时执行该语句块 --]
end

实例：

1564437356737

2.6.5 循环

学员完成

(1)while循环[==满足条件就循环==]

Lua 编程语言中 while 循环语句在判断条件为 true 时会重复执行循环体语句。
语法：

while(condition)
do
   statements
end

实例：

a=10
while( a < 20 )
do
   print("a 的值为:", a)
   a = a+1
end

效果如下：

1564437466576

(2)for循环

Lua 编程语言中 for 循环语句可以重复执行指定语句，重复次数可在 for 语句中控制。

语法： 1->10 1:exp1 10:exp2 2:exp3:递增的数量

for var=exp1,exp2,exp3 
do  
    <执行体>  
end

var 从 exp1 变化到 exp2，每次变化以 exp3 为步长递增 var，并执行一次 "执行体"。exp3 是可选的，如果不指定，默认为1。

例子：

for i=1,9,2
do
   print(i)
end

for i=1,9,2:i=1从1开始循环，9循环数据到9结束，2每次递增2

1564437629211

(3)repeat...until语句[==满足条件结束==]

Lua 编程语言中 repeat...until 循环语句不同于 for 和 while循环，for 和 while 循环的条件语句在当前循环执行开始时判断，而 repeat...until 循环的条件语句在当前循环结束后判断。

语法：

repeat
   statements
until( condition )

案例：

1564438640878

2.6.6 函数

lua中也可以定义函数，类似于java中的方法。例如：

--[[ 函数返回两个值的最大值 --]]
function max(num1, num2)

   if (num1 > num2) then
      result = num1;
   else
      result = num2;
   end

   return result; 
end
-- 调用函数
print("两值比较最大值为 ",max(10,4))
print("两值比较最大值为 ",max(5,6))

执行之后的结果：

两值比较最大值为     10
两值比较最大值为     6

..:表示拼接

2.6.7 表

table 是 Lua 的一种数据结构用来帮助我们创建不同的数据类型，如：数组、字典等。

Lua也是通过table来解决模块（module）、包（package）和对象（Object）的。

案例：

-- 初始化表
mytable = {}

-- 指定值
mytable[1]= "Lua"

-- 移除引用
mytable = nil

2.6.7 模块

(1)模块定义

模块类似于一个封装库，从 Lua 5.1 开始，Lua 加入了标准的模块管理机制，可以把一些公用的代码放在一个文件里，以 API 接口的形式在其他地方调用，有利于代码的重用和降低代码耦合度。

创建一个文件叫module.lua，在module.lua中创建一个独立的模块，代码如下：

module = {}

module.constant = "这是一个常量"

function module.func1()
    print("这是一个公有函数")
end

local function func2()
    print("这是一个私有函数！")
end

function module.func3()
    func2()
end

return module

由上可知，模块的结构就是一个 table 的结构，因此可以像操作调用 table 里的元素那样来操作调用模块里的常量或函数。

上面的 func2 声明为程序块的局部变量，即表示一个私有函数，因此是不能从外部访问模块里的这个私有函数，必须通过模块里的公有函数来调用.

(2)require 函数

require 用于引入其他的模块，类似于java中的类要引用别的类的效果。

用法：

require("<模块名>")

require "<模块名>"

两种都可以。

我们可以将上面定义的module模块引入使用,创建一个test_module.lua文件，代码如下：

-- test_module.lua 文件
-- module 模块为上文提到到 module.lua
require("module")

print(module.constant)

module.func3()

3 OpenResty介绍

OpenResty® 是一个基于 Nginx 与 Lua 的高性能 Web 平台，由中国人章亦春发起，其内部集成了大量精良的 Lua 库、第三方模块以及大多数的依赖项。用于方便地搭建能够处理超高并发、扩展性极高的动态 Web 应用、Web 服务和动态网关。

OpenResty® 通过汇聚各种设计精良的 Nginx 模块（主要由 OpenResty 团队自主开发），从而将 Nginx 有效地变成一个强大的通用 Web 应用平台。这样，Web 开发人员和系统工程师可以使用 Lua 脚本语言调动 Nginx 支持的各种 C 以及 Lua 模块，快速构造出足以胜任 10K 乃至 1000K 以上单机并发连接的高性能 Web 应用系统。

OpenResty® 的目标是让你的Web服务直接跑在 Nginx 服务内部，充分利用 Nginx 的非阻塞 I/O 模型，不仅仅对 HTTP 客户端请求,甚至于对远程后端诸如 MySQL、PostgreSQL、Memcached 以及 Redis 等都进行一致的高性能响应。

360，UPYUN，阿里云，新浪，腾讯网，去哪儿网，酷狗音乐等都是 OpenResty 的深度用户。

OpenResty 简单理解成就相当于封装了nginx,并且集成了LUA脚本，开发人员只需要简单的使用其提供了模块就可以实现相关的逻辑，而不再像之前，还需要在nginx中自己编写lua的脚本，再进行调用了。

3.1 安装openresty

linux安装openresty:

1.添加仓库执行命令

 yum install yum-utils
 yum-config-manager --add-repo https://openresty.org/package/centos/openresty.repo

2.执行安装

yum install openresty

3.安装成功后会在默认的目录如下：

/usr/local/openresty

3.2 安装nginx

默认已经安装好了nginx,在目录：/usr/local/openresty/nginx 下。

修改/usr/local/openresty/nginx/conf/nginx.conf,将配置文件使用的根设置为root,目的就是将来要使用lua脚本的时候，直接可以加载在root下的lua脚本。

cd /usr/local/openresty/nginx/conf
vi nginx.conf

修改代码如下：

1560739975500

3.3 测试访问

重启下centos虚拟机，然后访问测试Nginx

访问地址：http://192.168.220.110/

1560740292872

4.首页分类缓存处理

4.1 需求分析

需要在页面上显示商品分类的信息。

4.2 docker安装redis

使用docker安装redis

docker pull redis

docker run --name redis -p 6379:6379 -d redis

连接redis

4.3 Lua+Nginx配置

接下来我们使用lua来完成一个，操作nginx缓存, 访问redis,mysql数据库的案例

(1)Lua案例1需求：根据分类ID查询分类数据放入redis中

实现思路：

定义请求：用于查询数据库中的数据更新到redis中。
连接mysql ，按照分类ID读取商品分类，转换为json字符串。
连接redis，将品类json字符串存入redis 。

定义请求：

请求：
    /update_content
参数：
    id  --指定分类的id
返回值：
    json

请求地址：<http://192.168.220.110/update_content?id=1>

创建/root/lua目录，在该目录下创建update_content.lua：目的就是连接mysql 查询数据并存储到redis中。

1560759977349

上图代码如下：

ngx.header.content_type="application/json;charset=utf8"
local cjson = require("cjson")
local mysql = require("resty.mysql")
local uri_args = ngx.req.get_uri_args()
local id = uri_args["id"]

local db = mysql:new()
db:set_timeout(1000)
local props = {
   
    host = "192.168.220.110",
    port = 3306,
    database = "legou",
    user = "root",
    password = "root"
}

local res = db:connect(props)
local select_sql = "SELECT id_,is_parent_,order_,parent_id_,title_,expand_ FROM category_ WHERE id_= "..id
res = db:query(select_sql)
db:close()

local redis = require("resty.redis")
local red = redis:new()
red:set_timeout(2000)

local ip ="192.168.220.110"
local port = 6379
red:connect(ip,port)
red:set("content_"..id,cjson.encode(res))
red:close()

ngx.say("{flag:true}")

修改/usr/local/openresty/nginx/conf/nginx.conf文件：添加头信息，和 location信息

1560741056485

代码如下：

user root root;

server {
   
    listen       80;
    server_name  localhost;

    location /update_content {
   
        content_by_lua_file /root/lua/update_content.lua;
    }

        location /update_content {
   
                content_by_lua_file /root/lua/update_content.lua;
        }    

}

重启nginx

nginx -s reload

请求<http://192.168.220.110/update_content?id=1>可以实现缓存的添加`

1564422636804

(2)实现思路-从redis中获取数据

实现思路：

定义请求，用户根据分类的ID 获取的分类。通过lua脚本直接从redis中获取数据即可。

定义请求：

请求:/read_content
参数：id
返回值：json

在/root/lua目录下创建read_content.lua:

--设置响应头类型
ngx.header.content_type="application/json;charset=utf8"
--获取请求中的参数ID
local uri_args = ngx.req.get_uri_args()
local id = uri_args["id"]
--引入redis库
local redis = require("resty.redis")
--创建redis对象
local red = redis:new()
--设置超时时间
red:set_timeout(2000)
--连接
local ok, err = red:connect("192.168.220.110", 6379)
--获取key的值
local rescontent=red:get("content_"..id)
--输出到返回响应中
ngx.say(rescontent)
--关闭连接
red:close()

在/usr/local/openresty/nginx/conf/nginx.conf中配置如下：

如图：

1560741401873

代码：

location /read_content {
   
     content_by_lua_file /root/lua/read_content.lua;
}

访问http://192.168.220.110/read_content?id=1结果如下：

(3)加入openresty本地缓存

如上的方式没有问题，但是如果请求都到redis，redis压力也很大，所以我们一般采用多级缓存的方式来减少下游系统的服务压力。参考基本思路图的实现。

先查询openresty本地缓存如果没有

再查询redis中的数据，如果没有

再查询mysql中的数据，但凡有数据则返回即可。

修改read_content.lua文件，代码如下：

1560760965394

上图代码如下：

ngx.header.content_type="application/json;charset=utf8"
local uri_args = ngx.req.get_uri_args();
local id = uri_args["id"];
--获取本地缓存
local cache_ngx = ngx.shared.dis_cache;
--根据ID 获取本地缓存数据
local contentCache = cache_ngx:get('content_cache_'..id);

if contentCache == "" or contentCache == nil then
    local redis = require("resty.redis");
    local red = redis:new()
    red:set_timeout(2000)
    red:connect("192.168.220.110", 6379)
    local rescontent=red:get("content_"..id);

    if ngx.null == rescontent then
        local cjson = require("cjson");
        local mysql = require("resty.mysql");
        local db = mysql:new();
        db:set_timeout(2000)
        local props = {
   
            host = "192.168.220.110",
            port = 3306,
            database = "legou",
            user = "root",
            password = "root"
        }
        local res = db:connect(props);
    local select_sql = "SELECT id_,is_parent_,order_,parent_id_,title_,expand_ FROM category_ WHERE id_= "..id
        res = db:query(select_sql);
        local responsejson = cjson.encode(res);
        red:set("content_"..id,responsejson);
        ngx.say(responsejson);
        db:close()
    else
        cache_ngx:set('content_cache_'..id, rescontent, 10*60);
        ngx.say(rescontent)
    end
    red:close()
else
    ngx.say(contentCache)
end

定义lua缓存命名空间，修改nginx.conf，添加如下代码即可：

1560762051330

代码如下：

lua_shared_dict dis_cache 128m;

测试地址：http://192.168.220.110/read_content?id=1

此时会将分类ID=1的所有广告查询出来，并存入到Redis缓存。

1560761133343

测试地址：http://192.168.220.110/read_content?id=1

此时会获取分类ID=1的所有广告信息。

1560761192634

4.4 首页分类列表缓存方案

read_category.lua文件

ngx.header.content_type="application/json;charset=utf8"
local cache_ngx = ngx.shared.dis_cache;
local contentCache = cache_ngx:get("cl");

if contentCache == "" or contentCache == nil then
    local redis = require("resty.redis");
    local red = redis:new()
    red:set_timeout(2000)
    red:connect("192.168.220.110", 6379)
    local rescontent=red:get("cl");

    if ngx.null == rescontent then
        local cjson = require("cjson");
        local mysql = require("resty.mysql");
        local db = mysql:new();
        db:set_timeout(2000)
        local props = {
            host = "192.168.220.110",
            port = 3306,
            database = "legou",
            user = "root",
            password = "root"
        }
        local res = db:connect(props);
    local select_sql = "SELECT id_,is_parent_,order_,parent_id_,title_,expand_ FROM category_"
        res = db:query(select_sql);
        local responsejson = cjson.encode(res);
        red:set("cl",responsejson);
        ngx.say(responsejson);
        db:close()
    else
        cache_ngx:set("cl", rescontent, 10*60);
        ngx.say(rescontent)
    end
    red:close()
else
   ngx.say(contentCache)
end
-- 重定向到商品分类微服务
-- return ngx.redirect("http://192.168.220.1:8062/api/item/category/list");

nginx.conf

location /category/list {
        content_by_lua_file /root/lua/read_category.lua;
}

5 canal同步缓存数据

canal可以用来监控数据库数据的变化，从而获得新增数据，或者修改的数据。

阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务。

5.1 Canal工作原理

1560813843260

原理相对比较简单：

canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
mysql master收到dump请求，开始推送binary log给slave(也就是canal)
canal解析binary log对象(原始为byte流)

canal需要使用到mysql，我们需要先安装mysql,给大家发的虚拟机中已经安装了mysql容器，但canal是基于mysql的主从模式实现的，所以必须先开启binlog.

5.2 开启binlog模式

先使用docker 创建mysql容器,此处不再演示.

(1) 连接到mysql中,并修改/etc/mysql/mysql.conf.d/mysqld.cnf 需要开启主从模式，开启binlog模式。

执行如下命令，编辑mysql配置文件

1560814415655

命令行如下：

docker exec -it mysql /bin/bash
vi /etc/mysql/mysql.cnf

修改/etc/mysql/mysql.cnf配置文件，添加如下配置：

1560814236901

上图配置如下：

log-bin=/var/lib/mysql/mysql-bin
server-id=12345

(2) 创建账号用于测试使用,

使用root账号创建用户并授予权限

create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

(3)重启mysql容器

docker restart mysql

5.3 canal容器安装

下载镜像：

docker pull docker.io/canal/canal-server

容器安装

docker run -p 11111:11111 --name canal -d docker.io/canal/canal-server

进入容器,修改核心配置canal.properties 和instance.properties，canal.properties 是canal自身的配置，instance.properties是需要同步数据的数据库连接配置。

执行代码如下:

docker exec -it canal /bin/bash
cd canal-server/conf/
vi canal.properties
cd example/
vi instance.properties

修改canal.properties的id，不能和mysql的server-id重复，如下图：

1560814792482

修改instance.properties,配置数据库连接地址:

1560814968391

这里的canal.instance.filter.regex有多种配置，如下：

可以参考地址如下:

https://github.com/alibaba/canal/wiki/AdminGuide

mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\) 
常见例子：
1.  所有表：.*   or  .*\\..*
2.  canal schema下所有表： canal\\..*
3.  canal下的以canal打头的表：canal\\.canal.*
4.  canal schema下的一张表：canal.test1
5.  多个规则组合使用：canal\\..*,mysql.test1,mysql.test2 (逗号分隔)
注意：此过滤条件只针对row模式的数据有效(ps. mixed/statement因为不解析sql，所以无法准确提取tableName进行过滤)

配置完成后，设置开机启动，并记得重启canal。

docker update --restart=always canal
docker restart canal

5.4 canal微服务搭建

创建canal微服务工程，通过连接canal服务器，监控mysql的binlog，当mysql分类数据发生改变时，我们同步数据库数据到redis中，这样做到mysql和redis数据同步

(1)安装辅助jar包

在canal\spring-boot-starter-canal-master中有一个工程starter-canal，它主要提供了SpringBoot环境下canal的支持，我们需要先安装该工程，在starter-canal目录下执行mvn install，如下图：

1560815769342

(2)canal微服务工程搭建

legou-canal/pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>legou-parent</artifactId>
        <groupId>com.lxs</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>legou-canal</artifactId>


    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-config</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.retry</groupId>
            <artifactId>spring-retry</artifactId>
        </dependency>

        <!--canal依赖-->
        <dependency>
            <groupId>com.xpand</groupId>
            <artifactId>starter-canal</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>

        <dependency>
            <groupId>com.lxs</groupId>
            <artifactId>legou-item-instance</artifactId>
            <version>${project.version}</version>
        </dependency>

        <!-- redis 使用-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>


    </dependencies>


</project>

config-repo/canal-service.yml

server:
  port: 9007

logging:
  #file: demo.log
  pattern:
    console: "%d - %msg%n"
  level:
    org.springframework.web: debug
    com.lxs: debug

spring:
  redis:
    host: 192.168.220.110
    port: 6379
#canal配置
canal:
  client:
    instances:
      # exmaple
      example:
        host: 192.168.220.110
        port: 11111

legou-canal/src/main/resources/bootstrap.yml

spring:
  application:
    name: canal-service

(3)启动器

package com.lxs.legou.canal;


import com.xpand.starter.canal.annotation.EnableCanalClient;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
import org.springframework.cloud.openfeign.EnableFeignClients;


@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableEurekaClient
@EnableCanalClient
@EnableFeignClients
@EnableCircuitBreaker
public class CanalApplication {

    public static void main(String[] args) {
        SpringApplication.run(CanalApplication.class,args);
    }

}

(4)监听创建

创建一个CanalDataEventListener类，实现对表增删改操作的监听，代码如下：

package com.lxs.legou.canal.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lxs.legou.canal.client.CategoryClient;
import com.lxs.legou.item.po.Category;
import com.netflix.discovery.converters.Auto;
import com.xpand.starter.canal.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;

// 事件监听的注解 监听数据库的变化
@CanalEventListener
public class MyEventListener {
   

    //当数据被添加的时候触发
    // CanalEntry.EventType eventType  监听到的操作的类型  INSERT  UPDATE ,DELETE ,CREATE INDEX ,GRAND
    // CanalEntry.RowData rowData 被修改的数据()
    @InsertListenPoint
    public void onEvent(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
   
        //do something...
        System.out.println("添加数据监听。。。。");
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        for (CanalEntry.Column column : afterColumnsList) {
   
            System.out.println(column.getName()+":"+column.getValue());
        }
    }

    //当数据被更新的时候触发
    @UpdateListenPoint
    public void onEvent1(CanalEntry.RowData rowData) {
   
        //do something...
        System.out.println("修改数据监听。。。。");
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        for (CanalEntry.Column column : afterColumnsList) {
   
            System.out.println(column.getName()+":"+column.getValue());
        }
    }
    // 当数据被删除的时候触发
    @DeleteListenPoint
    public void onEvent3(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
   
        //do something...
        System.out.println("删除数据监听。。。。");
//        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        List<CanalEntry.Column> afterColumnsList = rowData.getBeforeColumnsList();
        for (CanalEntry.Column column : afterColumnsList) {
   
            System.out.println(column.getName()+":"+column.getValue());
        }
    }

    //自定义事件的触发
    // destination = "example" 指定某一个目的地 一定要和配置文件中的目录保持一致
    //schema = "canal-test" 要监听的数据库实例
    //table = {"t_user", "test_table"},   要监听的表
    // eventType = CanalEntry.EventType.UPDATE  要监听的类型
    @ListenPoint(destination = "example", schema = "legou", table = {
   "category_"}, eventType = {
   CanalEntry.EventType.UPDATE,CanalEntry.EventType.INSERT,CanalEntry.EventType.DELETE})
    public void onEvent4(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
   
        //do something...
        System.out.println("只监听分类。。。。");
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        for (CanalEntry.Column column : afterColumnsList) {
   
            System.out.println(column.getName()+":"+column.getValue());
        }
    }


}

(5)测试

启动canal微服务，然后修改任意数据库的表数据，canal微服务后台输出如下：

1560816240753

5.5 分类数据同步

1565065172531

如上图，每次执行分类操作的时候，会记录操作日志到，然后将操作日志发送给canal，canal将操作记录发送给canal微服务，canal微服务在同步最新的分类数据到redis中

在商品微服务中定义Feign调用接口

com.lxs.legou.item.api.CategoryApi

package com.lxs.legou.item.api;

import com.lxs.legou.core.json.JSON;
import com.lxs.legou.core.po.BaseEntity;
import com.lxs.legou.item.po.Category;
import io.swagger.annotations.ApiOperation;
import org.springframework.web.bind.annotation.*;

import java.util.List;

@RequestMapping("/category")
public interface CategoryApi {

    @ApiOperation(value="查询", notes="根据实体条件查询")
    @RequestMapping(value = "/list")
    public List<Category> list(Category category);

}

在canal微服务中创建Feign客户端接口，同时定义Hystix熔断方法

legou-canal/src/main/java/com/lxs/legou/canal/client/CategoryClient.java

package com.lxs.legou.canal.client;

import com.lxs.legou.item.api.CategoryApi;
import com.lxs.legou.item.po.Category;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestMapping;

import java.util.List;

@FeignClient(name = "item-service", fallback = CategoryClient.CategoryClientFallback.class)
public interface CategoryClient extends CategoryApi {


    @Component
    @RequestMapping("/category-fallback") //这个可以避免容器中requestMapping重复
    class CategoryClientFallback implements CategoryClient {

        private static final Logger LOGGER = LoggerFactory.getLogger(CategoryClientFallback.class);

        @Override
        public List<Category> list(Category category) {
            LOGGER.info("异常发生，进入fallback方法");
            return null;
        }
    }

}

legou-canal/src/main/java/com/lxs/legou/canal/listener/MyEventListener.java

package com.lxs.legou.canal.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lxs.legou.canal.client.CategoryClient;
import com.lxs.legou.item.po.Category;
import com.netflix.discovery.converters.Auto;
import com.xpand.starter.canal.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;

// 事件监听的注解 监听数据库的变化
@CanalEventListener
public class MyEventListener {

    @Autowired
    private CategoryClient categoryClient;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @ListenPoint(destination = "example", schema = "legou", table = {"category_"}, eventType = {CanalEntry.EventType.UPDATE,CanalEntry.EventType.INSERT,CanalEntry.EventType.DELETE})
    public void onEvent4(CanalEntry.EventType eventType, CanalEntry.RowData rowData) throws JsonProcessingException {
        List<Category> list = categoryClient.list(new Category());
        ObjectMapper objectMapper = new ObjectMapper();
        stringRedisTemplate.boundValueOps("category_list").set(objectMapper.writeValueAsString(list));
    }

}

通过MyEventListener监听器，通过canal监听category_数据变化是，同步分类数据到redis缓存

java电商项目（四）