Casbin分布式服务中如何使用Watcher观察者

2024-10-08 10:33:07 浏览数 (1)

概述

Casbin 是一个强大的、高效的开源访问控制框架，支持 Go, Java, Node.js, Javascript (React), Python, PHP, .NET, C , Rust 等十几种语言。

Casbin权限模型实战大揭秘，教育培训领域的创新实践和高效优化策略

观察者

通常在支持使用分布式消息系统，例如etcd保持多个Casbin执行器实例之间的一致性。因此，我们的用户可以同时使用多个Casbin 执行器来处理大量的权限检查请求。官方链接

原理

在常驻内存框架中使用Casbin。Swoole、Workerman、ReactPHP 运行模式为多进程，而多进程中数据是互相隔离的（每个进程都是独立互不干扰的，这意味着每个进程都维护着自己的资源、变量和类实例等）。注意：内存溢出，垃圾回收机制（GC）。

场景

当Enforcer中的策略发生变化时，调用 Watcher，向消息队列（MQ）中推动消息，监听该消息队列的Enforcer收到后，自动刷新该实例中的策略。

“注：在 PHP-FPM 环境下，并不需要Watcher，因为每个请求都是一个独立的fpm进程，都会实例化一个全新的Enforcer实例

实现

这里通过基于workerman的PHP异步redis客户端 workerman/redis 的发布订阅模式实现。workerman/redis 是基于workerman的异步redis组件。

发布订阅案例

接口实现

首先要实现官方的Watcher接口，同时实现接口的三个方法

代码语言：javascript复制
/**
 * @author Tinywan(ShaoBo Wan)
 * @date 2024/10/01 19:46
 */
declare(strict_types=1);

interface Watcher
{
    /**
     * Sets the callback function that the watcher will call when the policy in DB has been changed by other instances.
     * A classic callback is loadPolicy() method of Enforcer class.
     *
     * @param Closure $func
     */
    public function setUpdateCallback(Closure $func): void;

    /**
     * Update calls the update callback of other instances to synchronize their policy.
     * It is usually called after changing the policy in DB, like savePolicy() method of Enforcer class,
     * addPolicy(), removePolicy(), etc.
     */
    public function update(): void;

    /**
     * Close stops and releases the watcher, the callback function will not be called any more.
     */
    public function close(): void;
}

函数详解

  • 函数 setUpdateCallback(Closure $func) 传入一个匿名函数。设置当数据库中的策略被其他实例更改时，观察者将调用的回调函数。
  • 函数 update() 函数。调用其他实例的更新回调来同步它们的策略。它通常在更改 DB 中的策略后调用，如 Enforcer 类的 savePolicy() 方法。也是就添加策略会触发该函数（类似一个钩子吧！），这里会进行策略的发布，即 publish将信息发送到指定的频道（/casbin）
  • 函数 close() 函数。关闭停止并释放观察者，回调函数将不再被调用。

具体实现类

代码语言：javascript复制
/**
 * @author Tinywan(ShaoBo Wan)
 * @date 2024/10/01 19:46
 */
declare(strict_types=1);

use CasbinPersistWatcher;
use Closure;
use WorkermanRedisClient;

class RedisWatcher implements Watcher
{
    private Closure $callback;

    private $pubRedis;

    private $subRedis;

    private $channel;

    public function __construct(array $config)
    {
        $this->pubRedis = $this->createRedisClient($config);
        $this->subRedis = $this->createRedisClient($config);
        $this->channel = $config['channel'] ?? '/casbin';

        $this->subRedis->subscribe([$this->channel], function ($channel, $message) {
            if ($this->callback) {
                call_user_func($this->callback);
            }
        });
    }

    public function setUpdateCallback(Closure $func): void
    {
        $this->callback = $func;
    }

    public function update(): void
    {
        $this->pubRedis->publish($this->channel, 'casbin rules updated');
    }

    public function close(): void
    {
        $this->pubRedis->close();
        $this->subRedis->close();
    }

    private function createRedisClient(array $config): Client
    {
        $config['host'] = $config['host'] ?? '127.0.0.1';
        $config['port'] = $config['port'] ?? 6379;
        $config['password'] = $config['password'] ?? '';
        $config['database'] = $config['database'] ?? 0;

        $redis = new Client('redis://' . $config['host'] . ':' . $config['port']);
        $redis->auth($config['password'] ?? '');

        return $redis;
    }
}

使用

代码语言：javascript复制
/**
 * @author Tinywan(ShaoBo Wan)
 * @date 2024/10/01 19:46
 */
declare(strict_types=1);

use TinywanCasbinWatcherRedisWatcher;
use CasbinEnforcer;

$enforcer = new Enforcer("path/to/model.conf", "path/to/policy.csv");

$watcher = new RedisWatcher([
    'host' => '127.0.0.1',
    'password' => '123456',
    'port' => 6379,
    'database' => 0,
]);

// 设置监听者
$enforcer->setWatcher($watcher);

// 匿名回调加载策略
$watcher->setUpdateCallback(function () use ($enforcer){
    $enforcer->loadPolicy();
});
函数 接口 进程 权限 异步

0 人点赞