安装

sudo apt install shadowsocks-libev

配置

sudo vim /etc/shadowsocks-libev/config.json

作为客户端启动

ss-local -c /etc/shadowsocks-libev/config.json

后台启动

systemctl start shadowsocks-libev-local@config.service

参考
https://wiki.gulucat.com/doku.php/public/software/ss_install

<?php

namespace App\Http\Controllers;

use App\Service\IComputerService;
use App\Validator\IValidator;
use Illuminate\Http\Request;

class IndexController extends Controller
{
    private IComputerService $ComputerService;
    private IValidator $Validator;

    public function __construct(IComputerService $ComputerService, IValidator $Validator)
    {
        $this->ComputerService = $ComputerService;
        $this->Validator = $Validator;
    }

    public function index(Request $request)
    {
        $param = $this->Validator->check($request);

        $res = $this->ComputerService->index($param['per_page']);

        return response()->json($res);
    }
}

以上是一个控制器,

  private IComputerService $ComputerService; 

以上是业务逻辑层,限定IComputerService接口

  <?php

namespace App\Service;

interface IComputerService{
    public function index($per_page);
}

以上是一个IComputerService 业务逻辑层的接口

 public function __construct(IComputerService $ComputerService, IValidator $Validator)
    {
        $this->ComputerService = $ComputerService;
        $this->Validator = $Validator;
    }

以上是把$ComputerService注入到IndexController

那么在哪里自动注入到服务容器呢?

<?php

namespace App\Providers;

use App\Http\Controllers\IndexController;
use App\Service\IComputerService;
use App\Service\impl\ComputerServiceImpl;
use Illuminate\Support\ServiceProvider;

class AppServiceProvider extends ServiceProvider
{
    /**
     * Register any application services.
     */
    public function register(): void
    {
        //
    //    $this->app->bind(IComputerService::class, ComputerServiceImpl::class);

        $this->app->when(IndexController::class)
            ->needs(IComputerService::class)
            ->give(ComputerServiceImpl::class);
    }

    /**
     * Bootstrap any application services.
     */
    public function boot(): void
    {
        //
    }
}

以上代码是声明IndexController如果需要注入IComputerService接口,会自动调用ComputerServiceImpl实现类.

这样就可以实现代码解偶,符合依赖倒置原则,Dependency inversion principle,抽象不应依赖于细节，细节应依赖于抽象。

控制器如果要替换业务逻辑层即service层,不需要修改控制器的代码,只要新增和替换IComputerService 接口的实现类.

和在服务注册AppServiceProvider里修改一下 对应的实现类就可以了.

这里还有问题要注意.就是IComputerService没有指定返回数据的类型或者对象类型.这样不方便以后的修改或者替换.

以下是一个使用 RabbitMQ 的延迟队列来实现消息延迟发送的简单示例，使用 PHP 的 AMQP 扩展：

<?php
// 创建连接和channel
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 定义延迟队列和目标队列
$delayedQueueName = 'delayed_queue';
$targetQueueName = 'target_queue';

// 声明延迟队列
$channel->queue_declare($delayedQueueName, false, true, false, false, false, [
    'x-dead-letter-exchange' => '',
    'x-dead-letter-routing-key' => $targetQueueName
]);

// 声明目标队列
$channel->queue_declare($targetQueueName, false, true, false, false);

// 创建消息
$messageBody = 'Hello, delayed message!';
$message = new AMQPMessage($messageBody, [
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
]);

// 设置消息的 TTL 为 10 秒
$message->set('expiration', 10000);

// 发送消息到延迟队列
$channel->basic_publish($message, '', $delayedQueueName);

echo " [x] Sent 'Hello, delayed message!'\n";

// 关闭连接
$channel->close();
$connection->close();
?>

在这个示例中，我们创建了一个延迟队列 delayed_queue 和一个目标队列 target_queue。我们发送一条消息到延迟队列，并设置消息的 TTL 为 10 秒。当消息到达延迟时间后，会被投递到目标队列中。

请注意，以上示例仅为演示目的，实际应用中可能需要更多的错误处理和边界情况的考虑。

如何实现延时队列

https://juejin.cn/post/7094916704043712519

http://ai.devxiang.com/app/gpt

const http = require('http');

// 创建一个 HTTP 服务器
const server = http.createServer((req, res) => {
  // 设置响应头
  res.writeHead(200, {'Content-Type': 'application/json'});

  // 定义接口路径
  if (req.url === '/index') {
    // 模拟返回的数据
    const data = {
      message: 'Hello, this is the API response!'
    };

    var currentTime = new Date();

    data.message=currentTime;

    // 将数据转换为 JSON 格式并发送响应
    res.end(JSON.stringify(data));
  } else if (req.url === '/google') {
    const options = {
        hostname: 'cn.bing.com',
        path:'/HPImageArchive.aspx?format=js&idx=0&n=1',
        port: 80,
        method: 'GET',
        timeout:1000,
      };

    // 模拟返回的数据
     fetchDataFromServer(options, (error, data) => {
        if (error) {
            res.writeHead(200, {'Content-Type': 'application/json'});
            res.end(JSON.stringify(error));

        } else {
            const googleData = {
                message: data
              };
            res.writeHead(200, {'Content-Type': 'application/json'});
            res.end(JSON.stringify(googleData));

        }
      });

  } else {
    // 处理未知路径的请求
    res.writeHead(200, {'Content-Type': 'application/json'});
    res.end('Not Found');
  }
});

function fetchDataFromServer(options, callback) {
    const req = http.request(options, (res) => {
      let data = '';

      // 接收响应数据
      res.on('data', (chunk) => {
        data += chunk;
      });

      // 响应完成后处理数据
      res.on('end', () => {
        // 调用回调函数并传递获取的数据
        callback(null, data);
      });
    });

    // 处理请求错误
    req.on('error', (e) => {
      // 调用回调函数并传递错误信息
      callback(e, null);
    });

    // 结束请求
    req.end();
  }

// 监听端口
const port = 3000;
server.listen(port, () => {
  console.log(`Server is running on http://localhost:${port}`);
});

阻塞接口压测

wrk -c 100 -t 10 -d 15s http://127.0.0.1:3000/google

QPS几十

同时压测普通接口

wrk -c 100 -t 10 -d 10s http://127.0.0.1:3000/index

QPS 3万左右,没有明显影响.

index.js

var amqp = require('amqplib/callback_api');

const MyConsume = require('./MyConsume');

amqp.connect('amqp://name:password!@localhost:5672/vhost', function (error0, connection) {
    if (error0) {
        throw error0;
    }
    connection.createChannel(function (error1, channel) {
        if (error1) {
            throw error1;
        }

        var exchangeName = 'my_topic'; // 你的 exchange 名称
        var exchangeType = 'topic'; // exchange 类型，可以是 direct、fanout、topic 等
        var routingKey = 'my_routing_key'; // 你的 routing key
        var queueName = 'my_queue'; // 你的队列名称

        channel.assertExchange(exchangeName, exchangeType, { durable: true }, function(error2, ok) {
            if (error2) {
                throw error2;
            }
            console.log('Exchange ' + exchangeName + ' is ready');

            // 声明一个队列
            channel.assertQueue(queueName, { durable: true  }, function(error3, q) {
                if (error3) {
                    throw error3;
                }
                console.log('Queue ' +queueName + ' is created');

                // 将队列绑定到 exchange，并指定 routing key
                channel.bindQueue(queueName, exchangeName, routingKey);
                console.log('Queue ' + queueName + ' is bound to Exchange ' + exchangeName + ' with routing key ' + routingKey);

                // 在这里开始消费消息
                channel.consume(queueName, function(msg) {
                  MyConsume.handleMessage(msg,channel);
                }, {
                    noAck: false
                });
            });
        });
    });
});

MyConsume.js

const { UserModel } = require('./UserModel');

function handleMessage(msg,channel) {
    UserModel.create({
        user_name: 'Example Bookmark',
        url: 'http://www.example.com',
        type_id: 1,
        order: 1,
        is_recommend: true,
        status: true
      }).then(bookmark => {
        console.log('New bookmark created:');
        console.log(new Date());
      }).catch(error => {
        console.error('Error creating bookmark:', error);
      });
    // 在这里可以添加你的其他处理逻辑
    channel.ack(msg)
}

module.exports = {
    handleMessage: handleMessage
};

使用Mysql连接池,开10个mysql连接,消费3万rabbitmq消息,每条消息的逻辑是往mysql数据表写入一条数据,30秒写入完.

以上nodejs代码的写法我用php实现不到.哪怕是用swoole.如何用swoole来实现请高人指教.

php 代码

<?php

class Cat {
    private $age;
    private $color;

    public function __construct($age, $color) {
        $this->age = $age;
        $this->color = $color;
    }

    public function getAge() {
        return $this->age;
    }

    public function getColor() {
        return $this->color;
    }

    public function eat() {
        echo "Cat is eating.\n";
    }

    public function sleep() {
        echo "Cat is sleeping.\n";
    }
}

$myCat = new Cat(3, 'brown');
echo "Age: " . $myCat->getAge() . "\n";
echo "Color: " . $myCat->getColor() . "\n";
$myCat->eat();
$myCat->sleep();

?>

golang

package main

import "fmt"

// Cat 结构体
type Cat struct {
    age   int
    color string
}

func NewCat(age int, color string) *Cat {
    return &Cat{age, color}
}

func (c *Cat) GetAge() int {
    return c.age
}

func (c *Cat) GetColor() string {
    return c.color
}

func (c *Cat) Eat() {
    fmt.Println("Cat is eating.")
}

func (c *Cat) Sleep() {
    fmt.Println("Cat is sleeping.")
}

func main() {

    myCat := NewCat(3, "brown")
    fmt.Println("Age:", myCat.GetAge())
    fmt.Println("Color:", myCat.GetColor())
    myCat.Eat()
    myCat.Sleep()
}

日志
/var/www/java/xxl-job

删除文件

ubuntu停止rabbitmq

service rabbitmq-server stop

ubuntu启动rabbitmq

service rabbitmq-server start

queue队列持久化

  $channel->queue_declare($queue, false, true, false, false);

durable = true 队列持久化

message消息持久化

 $msg = new AMQPMessage('Hello World!'.$time, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));

重启 rabbitmq-server

queue和ready message依然存在