SpringCloud 请求多级缓存集群
简介
传统缓存的问题
传统的缓存策略一般是请求到达Tomcat后，先查询Redis，如果未命中则查询数据库，存在下面的问题：
- 请求要经过Tomcat处理，Tomcat的性能成为整个系统的瓶颈
- Redis缓存失效时，会对数据库产生冲击
多级缓存方案
多级缓存就是充分利用请求处理的每个环节，分别添加缓存，减轻Tomcat压力，提升服务性能：
JVM 进程缓存
缓存在日常开发中启动至关重要的作用，由于是存储在内存中，数据的读取速度是非常快的，能大量减少对数据库的访问，减少数据库的压力。我们把缓存分为两类：
- 分布式缓存，例如Redis：
- 优点：存储容量更大、可靠性更好、可以在集群间共享
- 缺点：访问缓存有网络开销
- 场景：缓存数据量较大、可靠性要求较高、需要在集群间共享
- 进程本地缓存，例如HashMap、GuavaCache：
- 优点：读取本地内存，没有网络开销，速度更快
- 缺点：存储容量有限、可靠性较低、无法共享
- 场景：性能要求较高，缓存数据量较小
本地缓存 Caffeine
Caffeine是一个基于Java8开发的，提供了近乎最佳命中率的高性能的本地缓存库。目前Spring内部的缓存使用的就是Caffeine。GitHub地址：https://github.com/ben-manes/caffeine
Caffeine 示例
@Test
void testBasicOps() {
// 创建缓存对象
Cache<String, String> cache = Caffeine.newBuilder().build();
// 存数据
cache.put("gf", "迪丽热巴");
// 取数据，不存在则返回null
String gf = cache.getIfPresent("gf");
System.out.println("gf = " + gf);
// 取数据，不存在则去数据库查询,key 为需要查询的值
String defaultGF = cache.get("defaultGF", key -> {
// 这里可以去数据库根据 key查询value
return "柳岩";
});
System.out.println("defaultGF = " + defaultGF);
}
Caffeine 缓存策略
Caffeine 作为一个缓存，自然就存在缓存过期，缓存满载等的处理策略，Caffeine 带有三种缓存策略，分别以下三种
- 基于容量驱逐缓存数据：设置缓存的数量上限，超出数量后，早前的缓存数据就会被清理
-
// 创建缓存对象 Cache<String, String> cache = Caffeine.newBuilder() .maximumSize(1) // 设置缓存大小上限为 1 .build(); - 基于过期时间：设置缓存的有效时间，超出时间的缓存将会被清理
-
// 创建缓存对象 Cache<String, String> cache = Caffeine.newBuilder() .expireAfterWrite(Duration.ofSeconds(10)) // 设置缓存有效期为 10 秒，从最后一次写入开始计时 .build(); - 基于引用使用GC清理：设置缓存为软引用或弱引用，利用GC回收缓存数据，性能差，不推荐
在默认情况下，当一个缓存元素过期的时候，Caffeine不会自动立即将其清理和驱逐。而是在一次读或写操作后，或者在空闲时间完成对失效数据的驱逐。
Caffeine 管单实现进程缓存案例
利用Caffeine实现下列需求：
- 给根据id查询商品的业务添加缓存，缓存未命中时查询数据库
- 给根据id查询商品库存的业务添加缓存，缓存未命中时查询数据库
- 缓存初始大小为100
- 缓存上限为10000
方案代码：
1.把Cache对象以Bean方式存入到IOC中
@Configuration
public class CaffeineConfig {
@Bean("itemCache")
public Cache<Long,String> itemCache(){
return Caffeine.newBuilder()
.initialCapacity(100) // 设置最小缓存空间大小
.maximumSize(10_000) // 设置最大缓存空间大小
.build();
}
}2.使用Cache的get方法，若缓存中不存在，则会调用Lambda中的获取数据，并写入缓存中，下次访问时，就不会再查询数据库了
@RestController
@RequestMapping("/caffeine")
public class TestController {
@Resource
private Cache<Long, String> itemCache;
@GetMapping("/get")
public String getCache() {
Long id = 1L;
String res = itemCache.get(id, key -> {
// 缓存中找不到的，通过查询数据库获得
return "database";
});
return res;
}
}
Nginx 配合集群 Tomcat 的缓存
我们知道 Caffeine 是一个进程缓存，它的缓存只能在当前Tomcat服务上能共享使用，而当我们的Tomcat使用了多台服务的集群负载均衡时，其它Tomcat服务器就无法获取到缓存了。
假设 /item/10001 的请求访问到Tomcat集群中时，Tomcat:8081 服务器读数据库后反回并使用Caffeine缓存数据
但当下一次访问 /item/10001 的请求时访问到 Tomcat:8082,因为 Caffeine 只能在 Tomcat:8081 中共享缓存，Tomcat:8082 无法获取，致使Tomcat:8082 又重复地访问数据存缓存。
解决方法：
我们可以使用Nginx的负载均衡策略 hash 来定义每一个请求应该指向访问哪台服务器
具体配置：
# tomcat集群配置
upstream tomcat-cluster{
hash $request_uri;
server 192.168.150.1:8081;
server 192.168.150.1:8082;
}
通过配置 hash $request_uri; 使每一条请求都会被计算成不同的hash值，只要请求uri不变，hash值就不变，hash值不变，哪么nginx则不会改变服务器指向，也就是说，使用hash值后，同一条请求永远只会被转发到同一台服务器中。
Nginx 业务集群开发
Nginx + Lua 开发业务
Lua 基本语法
lua 的基本类型：
// 空
nil
// 字符型
string
// 数字型
number
// 布尔型
boolean
// 方法型
function
// 表格型（数组和字典）
table
声明局部变量：
// 字符串
local str = 'xxx'
local str = "xxx"
// 数字
local num = 123.456
// 布尔
local boo = true
local boo = false
// table 索引
local arr = {"java", "c++", "python"}
// table map
local map = {name="jack", age= 21}
循环：
// 遍历数组型table
local arr = {"java", "python", "lua"}
for index,value in ipairs(arr) do
print(index, value)
end
// 遍历map型table
local map = {name="jack", age=21}
for key,value in pairs(map) do
print(key, value)
end
函数定义：
function 函数名(arg1, arg2 , ... , argn)
函数体
return 返回值
end
条件控制：
if(布尔表达式)
then
-- [ 布尔表达式为 true 时执行的代码块 ]
else
-- [ 布尔表达式为 flase 时执行的代码块 ]
end与java不同，布尔表达式中的逻辑运算是基于英文单词：
| 操作符 | 描述 | 实例 |
| and | 逻辑与操作符。 若 A 为 false，则返回 A，否则返回 B。 | (A and B) 为 false。 |
| or | 逻辑或操作符。 若 A 为 true，则返回 A，否则返回 B。 | (A or B) 为 true。 |
| not | 逻辑非操作符。与逻辑运算结果相反，如果条件为 true，逻辑非为 false。 | not(A and B) 为 true。 |
OpenResty
OpenResty® 是一个基于 Nginx的高性能 Web 平台，用于方便地搭建能够处理超高并发、扩展性极高的动态 Web 应用、Web 服务和动态网关。具备下列特点：
- 具备Nginx的完整功能(即OpenResty 包含Nginx系统)
- 基于Lua语言进行扩展，集成了大量精良的 Lua 库、第三方模块
- 允许使用Lua自定义业务逻辑、自定义库
官方网站： https://openresty.org/cn/
安装 OpenResty
1.安装开发依赖库
首先要安装OpenResty的依赖开发库，执行命令：
apt-get install libpcre3-dev \ libssl-dev perl make build-essential curl
CentOS
yum install pcre-devel openssl-devel gcc curl
2.安装OpenResty仓库
Ubuntu
你可以在你的 Ubuntu 系统中添加我们的 APT 仓库，这样就可以便于未来安装或更新我们的软件包（通过 apt-get update 命令）。 运行下面的命令就可以添加仓库（每个系统只需要运行一次）：
步骤一：安装导入 GPG 公钥时所需的几个依赖包（整个安装过程完成后可以随时删除它们）：
sudo apt-get -y install --no-install-recommends wget gnupg ca-certificates
步骤二：导入我们的 GPG 密钥：
- ubuntu 16 ~ 20 版本
wget -O - https://openresty.org/package/pubkey.gpg | sudo apt-key add - - ubuntu 22 及以上版本
wget -O - https://openresty.org/package/pubkey.gpg | sudo gpg --dearmor -o /usr/share/keyrings/openresty.gpg
步骤三：添加我们官方 APT 仓库。
对于 x86_64 或 amd64 系统，可以使用下面的命令：
- ubuntu 16 ~ 20 版本
echo "deb http://openresty.org/package/ubuntu $(lsb_release -sc) main" \ | sudo tee /etc/apt/sources.list.d/openresty.list - ubuntu 22 及以上版本
echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/openresty.gpg] http://openresty.org/package/ubuntu $(lsb_release -sc) main" | sudo tee /etc/apt/sources.list.d/openresty.list > /dev/null
而对于 arm64 或 aarch64 系统，则可以使用下面的命令:
- ubuntu 16 ~ 20 版本
echo "deb http://openresty.org/package/arm64/ubuntu $(lsb_release -sc) main" \ | sudo tee /etc/apt/sources.list.d/openresty.list - ubuntu 22 及以上版本
echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/openresty.gpg] http://openresty.org/package/arm64/ubuntu $(lsb_release -sc) main" | sudo tee /etc/apt/sources.list.d/openresty.list > /dev/null
步骤四：更新 APT 索引：
sudo apt-get update
然后就可以像下面这样安装软件包，比如 openresty：
sudo apt-get -y install openresty
这个包同时也推荐安装 openresty-opm 和 openresty-restydoc 包，所以后面两个包会缺省安装上。 如果你不想自动关联安装，可以用下面方法关闭自动关联安装：
sudo apt-get -y install --no-install-recommends openresty
CentOS
你可以在你的 CentOS 系统中添加 openresty 仓库，这样就可以便于未来安装或更新我们的软件包（通过 yum check-update 命令）。 运行下面的命令就可以添加我们的仓库（对于 CentOS 8 或以上版本，应将下面的 yum 都替换成 dnf）：
CentOS 9 或者更新版本
# add the yum repo:
wget https://openresty.org/package/centos/openresty2.repo
sudo mv openresty2.repo /etc/yum.repos.d/openresty.repo
# update the yum index:
sudo yum check-update
CentOS 8 或者更老版本
# add the yum repo:
wget https://openresty.org/package/centos/openresty.repo
sudo mv openresty.repo /etc/yum.repos.d/openresty.repo
# update the yum index:
sudo yum check-update
然后就可以像下面这样安装软件包，比如 openresty：
sudo yum install -y openresty
如果你想安装命令行工具 resty，那么可以像下面这样安装 openresty-resty 包：
sudo yum install -y openresty-resty
命令行工具 opm 在 openresty-opm 包里，而 restydoc 工具在 openresty-doc 包里头。
列出所有 openresty 仓库里头的软件包：
sudo yum --disablerepo="*" --enablerepo="openresty" list available
3.安装opm工具
opm是OpenResty的一个管理工具，可以帮助我们安装一个第三方的Lua模块。
如果你想安装命令行工具 opm，那么可以像下面这样安装 openresty-opm 包：
apy-get install -y openresty-opm
4.目录结构
默认情况下，OpenResty安装的目录是：/usr/local/openresty
看到里面的nginx目录了吗，OpenResty就是在Nginx基础上集成了一些Lua模块。
5.配置nginx的环境变量
目的是为了在任意目录下都能运行nginx命令
打开配置文件：
vi /etc/profile在最下面加入两行：
export NGINX_HOME=/usr/local/openresty/nginx
export PATH=${NGINX_HOME}/sbin:$PATHNGINX_HOME：后面是OpenResty安装目录下的nginx的目录
然后让配置生效：
source /etc/profile
运行OpenResty
OpenResty底层是基于Nginx的，查看OpenResty目录的nginx目录，结构与windows中安装的nginx基本一致：
所以运行方式与nginx基本一致：
# 启动nginx
nginx
# 重新加载配置
nginx -s reload
# 停止
nginx -s stop
nginx的默认配置文件注释太多，影响后续我们的编辑，这里将nginx.conf中的注释部分删除，保留有效部分。
#user nobody;
worker_processes 1;
error_log logs/error.log;
events {
worker_connections 1024;
}
http {
include mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
server {
listen 8081;
server_name localhost;
location / {
root html;
index index.html index.htm;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
}
}使用命令 nginx 即可启动。
使用OpenResty与Lua实现请求处理
OpenResty 可以对请求进行拦截并处理，可以配合Lua来处理这些请求，这样就可以做到不需要经过Tomcat等服务器就能拦截请求并响应数据。
做这样的工作是为了可以在OpenResty中配合Lua来访问Redis查询缓存，就无需到达Tomcat后再访问Redis了。
步骤：
- 1.在nginx.conf的http下面，添加对OpenResty的Lua模块的加载（固定写法）：
-
# 加载lua 模块 lua_package_path "/usr/local/openresty/lualib/?.lua;;"; # 加载c模块 lua_package_cpath "/usr/local/openresty/lualib/?.so;;"; - 2.在nginx.conf的server下面，添加对/api/item这个路径的监听：
-
location ~ /api/item/(\d+) { # ~ 代表使用正则表达式对请求路径进行匹配 # 响应类型，这里返回json default_type application/json; # 响应数据由 lua/item.lua这个文件来决定 content_by_lua_file lua/item.lua; }这里使用了 content_by_lua_file 指向了文件 lua/item.lua，这个文件通常是访问在 nginx 根目录下的
- 3.在Lua文件中，初步尝试返回一些json响应数据，使用 ngx.say() 方法：
-
-- 返回假数据，这里的ngx.say()函数，就是写数据到Response中 ngx.say('{"id":10001,"name":"SALSA AIR}')这样当访问 /api/item 的路径时，NG会调用Lua文件作为响应，而Lua文件返回的Response为一个json，则前端会收到对应的json数据，实现OpenResty与Lua配合拦截请求。
- 下一步我们可以使用Lua控制读取Redis获取缓存。
OpenResty 与 Lua 获取请求参数
OpenResty提供了各种API用来获取不同类型的请求参数：
| 参数格式 | 参数示例 | 参数解析代码示例 |
| 路径占位符 | /item/1001 | |
| 请求头 | id：1001 | -- 获取请求头，返回值是table类型
local headers = ngx.req.get_headers() |
| Get请求参数 | ?id=1001 | -- 获取GET请求参数，返回值是table类型
local getParams = ngx.req.get_uri_args() |
| Post表单参数 | id=1001 | -- 读取请求体
ngx.req.read_body() -- 获取POST表单参数，返回值是table类型 local postParams = ngx.req.get_post_args() |
| JSON参数 | {"id": 1001} | -- 读取请求体
ngx.req.read_body() -- 获取body中的json参数，返回值是string类型 local jsonBody = ngx.req.get_body_data() |
OpenResty 与 Lua 向 Tomcat 发送请求获取数据
满足下面的需求：
- 1.获取请求参数中的id（上一章节中我们知道如何获取请求参数了）
- 2.根据id向Tomcat服务发送请求，查询商品信息
- 3.根据id向Tomcat服务发送请求，查询库存信息（与2.基本相同）
- 4.组装商品信息、库存信息，序列化为JSON格式并返回
实现概念：
Nginx给Lua提供了内部API方法，让Lua发送Http请求：
local resp = ngx.location.capture("/请求路径",{ 请求参数 })
其中请求参数可以使用以下三种：
method = ngx.HTTP_GET, -- 请求方式
args = {a=1,b=2}, -- get方式传参数
body = "c=3&d=4" -- post方式传参数
其中 resp 返回的响应内容包括：
- resp.status：响应状态码
- resp.header：响应头，是一个table
- resp.body：响应体，就是响应数据
基础实现：
local resp = ngx.location.capture("/path",{
method = ngx.HTTP_GET, -- 使用GET请求方式
args = {a=1,b=2} -- get方式传参数
})
解析：Lua使 OpenResty 让 Nginx 使用 GET 方法访问 /path 路径的请求，请求的参数为 a=1&b=2
请求路径转发：
在上面实现中，我们使用了 ngx.location.capture 让lua实现请求，但我集团发现，这个请求路径仅仅只有一个路径，没有包含服务器地址和端口等信息。
这其是因为，/path 的请求路径会被 nginx 所捕获，若我们的Tomcat服务器是在另一台服务器上的，那么我们需要在OpenResty中的Nginx配置中另外做一个反向代理服务器请求：
#user nobody;
worker_processes 1;
error_log logs/error.log;
events {
worker_connections 1024;
}
http {
include mime.types;
default_type application/octet-stream;
sendfile on;
keepalive_timeout 65;
server {
listen 8081;
server_name localhost;
# 对正则的请求进行配置
location ~ api/item/(\d+) {
# 使用json作为响应类型
default_type application/json;
# 响应结果由 lua/item.lua 文件来决定
content_by_lua_file lua/item.lua
}
# 对Lua中的请求路径开头的应设定转发到Tomcat服务器地址上
location /path {
proxy_pass http://192.168.xx.xx:8080
}
location / {
root html;
index index.html index.htm;
}
error_page 500 502 503 504 /50x.html;
location = /50x.html {
root html;
}
}
}
优化：
为了使 ngx.location.capture 方法复用，我们可以把调用这个方法进行抽取，并封装成一个工具方法，随时让其它Lua代码调用：
-- 封装函数，发送http请求，接收请求路径和请求参数,并解析响应
local function read_http(path, params)
local resp = ngx.location.capture(path,{
method = ngx.HTTP_GET,
args = params,
})
-- 请求结果为 nil 时
if not resp then
-- 记录错误信息，返回404
ngx.log(ngx.ERR, "http not found, path: ", path , ", args: ", args)
ngx.exit(404)
end
return resp.body
end
-- 将方法导出，相当于把方法封装到一个table中，并把这个table导出去
local _M = {
read_http = read_http
}
return _M
这个文件可以存放到 openresty/lualib/common.lua 目录中，因为我们上面在 nginx 中配置好了以下代码：
# 加载lua 模块
lua_package_path "/usr/local/openresty/lualib/?.lua;;"; 因此在nginx中，都能访问到这个文件中的导出方法。
Lua 调用封装的请求方法获取Tomcat数据：
通过上面的OpenResty配置后，我们可以在 item.lua 上编写lua代码，以在请求时使用lua来控制获取Tomcat数据
-- 引入自定义工具模块
local common = require("common")
local read_http = common.read_http
-- 获取路径参数
local id = ngx.var[1]
-- 根据id查询商品
local itemJSON = read_http("/item/".. id, nil)
-- 根据id查询商品库存
local itemStockJSON = read_http("/item/stock/".. id, nil)
当我们获取到Tomcat的数据后，我们需要对数据进行整合，这里需要把json数据反序列化为table数据，用到Nginx提供的 cjson 库工具：
OpenResty提供了一个cjson的模块用来处理JSON的序列化和反序列化。
官方地址： https://github.com/openresty/lua-cjson/
- 引入cjson模块：
-
local cjson = require "cjson" - 序列化：
-
local obj = { name = 'jack', age = 21 } local json = cjson.encode(obj) - 反序列化：
-
local json = '{"name": "jack", "age": 21}' -- 反序列化 local obj = cjson.decode(json); print(obj.name)
最终我们可以利用cjson把json数据转为lua的table类型数据，并进行整合后响应：
-- 引入自定义工具模块
local common = require("common")
local read_http = common.read_http
-- 引入cjson工具模块
local cjson = require("cjson")
-- 获取路径参数
local id = ngx.var[1]
-- 根据id查询商品
local itemJSON = read_http("/item/".. id, nil)
-- 根据id查询商品库存
local itemStockJSON = read_http("/item/stock/".. id, nil)
-- 把两个json数据反序列化为table
local item = cjson.decode(itemJSON)
local stock = cjson.decodel(itemStockJSON)
-- 把stock 中的数据合并到 item中
item.stock = stock.stock
item.sold = stock.sold
-- 把合并后的数据序列化为json并响应
ngx.say(cjson.encode(item))
搭建 Redis 作为缓存
从上面我们知道，Redis可以作为Nginx的缓存，这样就可以不需要经过Tomcat直接获取到Redis缓存数据
数据预热
如果我们启动服务后，等待用户访问时，我们再去查询数据库存到Redis的话，对于高并发网站而言，启动时的瞬间大量访问有可能会使数据库压力过大，所以我们可以在服务启动时，先把一些访问量比较大的数据预先存入Redis中，这样可以有效防止服务启动后的空Redis缓存而带来的大量数据库查询问题：
我们在SpringBoot项目中创建一个类，实现InitializingBean 接口在Bean装载完成后，对Redis缓存热点数据
@Component
public class RedisHandler implements InitializingBean {
@Autowired
private StringRedisTemplate redisTemplate;
@Override
public void afterPropertiesSet() throws Exception {
// 查询数据库获取相比执门的数据
// 使用 redisTemplate 把这些数据预先写入到Redis中缓存着
}
}
OpenResty 配合 Lua 优先读 Redis 中的缓存
我们在访问时，应该在Lua中编写逻辑，让请求优先访问Redis中的数据，若查询失败，我们再去查询Tomcat
Lua使用Redis客户端
OpenResty 提供了操作Redis的模块，在需要操作Redis时，只要引入该模块就可以。
-- 引入redis模块
local redis = require("resty.redis")
-- 初始化Redis对象
local red = redis:new()
-- 设置Redis超时时间（建立连接超时，发送请求超时，Redis响应超时）
red:set_timeouts(1000, 1000, 1000)
封装用于读取Redis的函数，和关闭连接的函数
-- 查询redis的方法 ip和port是redis地址，key是查询的key
local function read_redis(ip, port, key)
-- 获取一个连接
local ok, err = red:connect(ip, port)
if not ok then
ngx.log(ngx.ERR, "连接redis失败 : ", err)
return nil
end
-- 查询redis
local resp, err = red:get(key)
-- 查询失败处理
if not resp then
ngx.log(ngx.ERR, "查询Redis失败: ", err, ", key = " , key)
end
--得到的数据为空处理
if resp == ngx.null then
resp = nil
ngx.log(ngx.ERR, "查询Redis数据为空, key = ", key)
end
close_redis(red)
return resp
end
-- 关闭redis连接的工具方法，其实是放入连接池
local function close_redis(red)
local pool_max_idle_time = 10000 -- 连接的空闲时间，单位是毫秒
local pool_size = 100 --连接池大小
local ok, err = red:set_keepalive(pool_max_idle_time, pool_size)
if not ok then
ngx.log(ngx.ERR, "放入Redis连接池失败: ", err)
end
end
最终把代码放到 common.lua 工具中
-- 引入redis模块
local redis = require("resty.redis")
-- 初始化Redis对象
local red = redis:new()
-- 设置Redis超时时间（建立连接超时，发送请求超时，Redis响应超时）
red:set_timeouts(1000, 1000, 1000)
-- 查询redis的方法 ip和port是redis地址，key是查询的key
local function read_redis(ip, port, key)
-- 获取一个连接
local ok, err = red:connect(ip, port)
if not ok then
ngx.log(ngx.ERR, "连接redis失败 : ", err)
return nil
end
-- 查询redis
local resp, err = red:get(key)
-- 查询失败处理
if not resp then
ngx.log(ngx.ERR, "查询Redis失败: ", err, ", key = " , key)
end
--得到的数据为空处理
if resp == ngx.null then
resp = nil
ngx.log(ngx.ERR, "查询Redis数据为空, key = ", key)
end
-- 关闭Redis查询连接
close_redis(red)
return resp
end
-- 关闭redis连接的工具方法，其实是放入连接池
local function close_redis(red)
local pool_max_idle_time = 10000 -- 连接的空闲时间，单位是毫秒
local pool_size = 100 --连接池大小
local ok, err = red:set_keepalive(pool_max_idle_time, pool_size)
if not ok then
ngx.log(ngx.ERR, "放入Redis连接池失败: ", err)
end
end
-- 封装函数，发送http请求，接收请求路径和请求参数,并解析响应
local function read_http(path, params)
local resp = ngx.location.capture(path,{
method = ngx.HTTP_GET,
args = params,
})
-- 请求结果为 nil 时
if not resp then
-- 记录错误信息，返回404
ngx.log(ngx.ERR, "http not found, path: ", path , ", args: ", args)
ngx.exit(404)
end
return resp.body
end
-- 将方法导出，相当于把方法封装到一个table中，并把这个table导出去
local _M = {
read_http = read_http,
read_redis = read_redis
}
return _M
接下来我们需要改造Nginx请求的lua逻辑方法：
-- 引入自定义工具模块
local common = require("common")
local read_http = common.read_http
local read_redis = common.read_redis -- 引入工具类中的读Redis方法
-- 引入cjson工具模块
local cjson = require("cjson")
-- 获取路径参数
local id = ngx.var[1]
-- 定义一个方法，用于处理访问时优先访问Redis,再访问Tomcat
function read_data(key, path, param)
local resp = read_redis("127.0.0.1", 6379, key)
if not resp then -- 如果redis查询没有数据，则向Tomcat中获取数据
-- 查询Tomcat
resp = read_http(path, param)
end
return resp
end
-- 根据id查询商品,不再直接访问 Tomcat 而是调用 read_data 方法
local itemJSON = read_data("item:id" .. id ,"/item/".. id, nil)
-- 根据id查询商品库存,不再直接访问 Tomcat 而是调用 read_data 方法
local itemStockJSON = read_data("item:stock:id" .. id ,"/item/stock/".. id, nil)
-- 把两个json数据反序列化为table
local item = cjson.decode(itemJSON)
local stock = cjson.decodel(itemStockJSON)
-- 把stock 中的数据合并到 item中
item.stock = stock.stock
item.sold = stock.sold
-- 把合并后的数据序列化为json并响应
ngx.say(cjson.encode(item))
OpenResty 本地缓存
OpenResty为Nginx提供了shard dict的功能，可以在nginx的多个worker之间共享数据，实现缓存功能。
- 开启共享字典，在nginx.conf的http下添加配置：
-
http { # 共享字典，也就是本地缓存，名称叫做：item_cache，大小150m lua_shared_dict item_cache 150m; } - 操作共享字典：
-
-- 获取本地缓存对象 local item_cache = ngx.shared.item_cache -- 存储, 指定key、value、过期时间，单位s，默认为0代表永不过期 item_cache:set('key', 'value', 1000) -- 读取 local val = item_cache:get('key')
然后我们对lua逻辑进行修改：
-- 引入自定义工具模块
local common = require("common")
local read_http = common.read_http
local read_redis = common.read_redis -- 引入工具类中的读Redis方法
-- 引入cjson工具模块
local cjson = require("cjson")
-- 获取路径参数
local id = ngx.var[1]
-- 获取本地缓存对象
local item_cache = ngx.shared.item_cache
-- 定义一个函数，先查询本地缓存，再查询redis，再查询http
local function read_data(key, expire, path, params)
-- 先读取本地缓存
local val = item_cache:get(key)
if not val then
-- 缓存未命中，记录日志
ngx.log(ngx.ERR, "本地缓存查询失败, key: ", key , ", 尝试redis查询")
-- 查询redis
val = read_redis("127.0.0.1", 6379, key)
-- 判断redis是否命中，如果没有命中，再去查询Tomcat
if not val then
ngx.log(ngx.ERR, "Redis缓存查询失败, key: ", key , ", 尝试http查询")
-- Redis查询失败，查询http
val = read_http(path, params)
end
end
-- 查询完成后，再写入本地缓存
item_cache:set(key, val, expire)
return val
end
-- 根据id查询商品,不再直接访问 Tomcat 而是调用 read_data 方法
local itemJSON = read_data("item:id" .. id ,"/item/".. id, nil)
-- 根据id查询商品库存,不再直接访问 Tomcat 而是调用 read_data 方法
local itemStockJSON = read_data("item:stock:id" .. id ,"/item/stock/".. id, nil)
-- 把两个json数据反序列化为table
local item = cjson.decode(itemJSON)
local stock = cjson.decodel(itemStockJSON)
-- 把stock 中的数据合并到 item中
item.stock = stock.stock
item.sold = stock.sold
-- 把合并后的数据序列化为json并响应
ngx.say(cjson.encode(item))到此做到了以下图中的效果：
- 用户访问时
- 1.先访问OpenResty中的缓存（共享字典）
- 2.访问Redis中的数据
- 3.访问Caffeine中的数据
- 4.访问MySQL数据库中的数据
Canal 基于数据库监听实现数据同步
当我们的web系统增加了Redis的缓存后，若我们的数据库数据发生变化时，缓存应如何同步数据库中的数据呢？
缓存数据同步的常见方式有三种：
- 设置有效期（超时删除）：给缓存设置有效期，到期后自动删除。再次查询时更新
- 优势：简单、方便
- 缺点：时效性差，缓存过期之前可能不一致
- 场景：更新频率较低，时效性要求低的业务
- 同步双写（在增删改代码中同步）：在修改数据库的同时，直接修改缓存
- 优势：时效性强，缓存与数据库强一致
- 缺点：有代码侵入，耦合度高；
- 场景：对一致性、时效性要求较高的缓存数据
- 异步通知（RabbitMQ方案）：修改数据库时发送事件通知，相关服务监听到通知后修改缓存数据
- 优势：低耦合，可以同时通知多个缓存服务
- 缺点：时效性一般，可能存在中间不一致状态
- 场景：时效性要求一般，有多个服务需要同步
RabbitMQ 的数据同步慨念
Canal 的数据同步慨念
Canal 简介
Canal [kə'næl]，译意为水道/管道/沟渠，canal是阿里巴巴旗下的一款开源项目，基于Java开发。基于数据库增量日志解析，提供增量数据订阅&消费。GitHub的地址：https://github.com/alibaba/canal
Canal是基于mysql的主从同步来实现的，MySQL主从同步的原理如下：
- MySQL master 将数据变更写入二进制日志( binary log），其中记录的数据叫做binary log events
- MySQL slave 将 master 的 binary log events拷贝到它的中继日志(relay log)
- MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
Canal就是把自己伪装成MySQL的一个slave节点，从而监听master的binary log变化。再把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。
安装 Canal
要安装Canal之前，需要对MySQL开启主从关系功能：
1.MySQL开启主从功能（开启Binlog）：
修改Mysql的配置文件my.cnf,添加Binlog相关配置
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=unsoft配置解读：
log-bin=/var/lib/mysql/mysql-bin：设置binary log文件的存放地址和文件名，叫做mysql-binbinlog-do-db=unsoft：指定对哪个database记录binary log events，这里记录heima这个库
最终配置：
[mysqld]
skip-name-resolve # 跳过DNS解析
character_set_server=utf8 # 设置UTF8编码
datadir=/var/lib/mysql # 数据存放地址
server-id=1000 # 进程id设为1000
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=unsoft然后重启mysql服务器
2.创建用户并设置用户权限
接下来添加一个仅用于数据同步的账户，出于安全考虑，这里仅提供对heima这个库的操作权限。
create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
FLUSH PRIVILEGES;
测试设置是否成功配置Binlog：在mysql控制台，或者Navicat中，输入命令：
show master status;
3.配置Docker网络
docker 配置网络（如果使用docker的话）
docker network create mysql-network让mysql加入到这个网络中：
docker network connect heima mysql
4.Docker 运行 Canal
运行命令：
docker run -p 11111:11111 --name canal \
-e canal.destinations=canal-colony \
-e canal.instance.master.address=mysql:3306 \
-e canal.instance.dbUsername=canal \
-e canal.instance.dbPassword=canal \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false \
-e canal.instance.filter.regex=unsoft\\..* \
--network mysql-network \
-d canal/canal-server:v1.1.5说明:
-p 11111:11111：这是canal的默认监听端口-e canal.destinations=canal-colony：canal支持集群，这里是声明一个canal集群实例名称-e canal.instance.master.address=mysql:3306：数据库地址和端口，如果不知道mysql容器地址，可以通过docker inspect 容器id来查看-e canal.instance.dbUsername=canal：数据库用户名-e canal.instance.dbPassword=canal：数据库密码-e canal.instance.filter.regex=：要监听的表名称
表名称监听支持的语法：
mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\)
常见例子：
1. 所有表：.* or .*\\..*
2. canal schema下所有表： canal\\..*
3. canal下的以canal打头的表：canal\\.canal.*
4. canal schema下的一张表：canal.test1
5. 多个规则组合使用然后以逗号隔开：canal\\..*,mysql.test1,mysql.test2 运行canal完成。
使用Canal客户端同步数据
Canal提供了各种语言的客户端，当Canal监听到binlog变化时，会通知Canal的客户端。
Canal 通过监听数据库的变化后，不会自动操作缓存，需要我们自定义操作，因此我们在Java项目中需要引入Canal客户端，再做对应的缓存更新操作。
Canal提供了各种语言的客户端，当Canal监听到binlog变化时，会通知Canal的客户端。不过这里我们会使用GitHub上的第三方开源的canal-starter。地址：https://github.com/NormanGyllenhaal/canal-client
引入依赖
<!--canal-->
<dependency>
<groupId>top.javatool</groupId>
<artifactId>canal-spring-boot-starter</artifactId>
<version>1.2.1-RELEASE</version>
</dependency>
yaml 配置文件中配置Canal服务器地址
canal:
destination: canal-colony # canal实例名称，要跟canal-server运行时设置的destination一致
server: 192.168.150.101:11111 # canal地址
代码实现：我们需要实现Canal的一个接口【EntryHandler<>】中的三个方法：
- 1.需要定义监听哪个表，使用 @CanalTable("") 注解声明
- 2.Canal 会自动把表中的发生变化的数据转成实体类，所以需要提供实体类泛型
@CanalTable("tb_item")
@Component
public class ItemHandler implements EntryHandler<Item> {
// 当表中存在增加数据时
@Override
public void insert(Item item) {
// 新增数据到redis
}
// 当表中存在修改数据时
@Override
public void update(Item before, Item after) {
// 更新redis数据
// 更新本地缓存
}
// 当表中存在删除数据时
@Override
public void delete(Item item) {
// 删除redis数据
// 清理本地缓存
}
}
因为 Canal 会自动对表中发生改变的数据转成实体类，因此它需要对实体类中的某些不能识别的字段做特别的标记，因为Canal并不依赖Mybatis，所以配置实体类中的字段兼容Canal需要特别配置：
@Data
@TableName("tb_item")
public class Item {
@TableId(type = IdType.AUTO)
@Id
private Long id;
@Column(name = "name")
private String name;
// ... 其它字段略
private Date updateTime;
@TableField(exist = false)
@Transient
private Integer stock;
@TableField(exist = false)
@Transient
private Integer sold;
}
解析：
- Canal 需要指定实体类中表示主键的成员变量，使用 @Id 注解声明。
- Canal 支持字段转成员变量时改为驼峰命名，但如果成员变量无法和字段对应时，可以使用 @Column(name = "") 字段声明。
- 如果实体类中存在与字段不相关的字段（不存在表中的字段），可以使用 @Transient 字段声明
共有 0 条评论