Reactor模式是一种处理并发请求的事件驱动设计模式，常用于高性能网络编程。它基于I/O多路复用机制，使用同步非阻塞的方式处理客户端请求。下面我将详细介绍Reactor模式的结构和工作原理。

## Reactor模式的结构
Reactor模式由以下几个角色构成：

1. Reactor（反应器）：负责监听并分发事件，它是整个模式的核心。Reactor会注册和管理所有的事件处理器，并在事件到达时将其分发给对应的处理器。



2. Handlers（处理器）：负责具体的事件处理逻辑。每个处理器都会注册到Reactor中，当事件到达时，Reactor会将事件分发给对应的处理器进行处理。

3. Event（事件）：代表一个客户端请求或其他需要处理的事件。事件可以是输入事件（如客户端连接请求）或输出事件（如数据发送完成）。

4. Demultiplexer（多路复用器）：负责监听事件的到达，并将事件分发给Reactor。常见的多路复用器有select、poll和epoll等。

5. Dispatcher（事件分发器）：负责将事件分发给对应的处理器。它可以是同步的，也可以是异步的。

## Reactor模式的工作原理
Reactor模式的工作流程如下：

1. Reactor启动并初始化，开始监听事件。

2. 当有事件到达时，多路复用器会通知Reactor。

3. Reactor接收到事件后，根据事件类型选择合适的处理器进行处理。

4. 处理器执行相应的业务逻辑，完成后返回结果。

5. Reactor将处理结果返回给客户端或进行下一步处理。

6. 重复上述步骤，处理更多的事件。

## Reactor模式的优点
- 高性能：Reactor模式使用了同步非阻塞的方式处理事件，能够充分利用CPU资源，提高系统的并发处理能力。

- 可扩展性：通过添加更多的处理器，可以轻松地扩展系统的处理能力，适应不断增长的并发请求。

- 灵活性：Reactor模式将事件的处理与具体的业务逻辑分离，使得系统更加灵活，易于维护和扩展。

## Reactor模式的适用场景
Reactor模式适用于以下场景：

- 高并发的网络编程：当系统需要处理大量的并发请求时，Reactor模式能够提供高性能的解决方案。

- 长连接的通信：如果系统中存在长时间保持连接的情况，并且通信频率不高，Reactor模式可以有效地管理连接和处理事件。

## 总结
Reactor模式是一种处理并发请求的事件驱动设计模式，通过使用同步非阻塞的方式处理事件，提高系统的并发处理能力。它由Reactor、处理器、事件、多路复用器和事件分发器等组成，具有高性能、可扩展性和灵活性的优点。在高并发的网络编程和长连接通信等场景下，Reactor模式是一个值得考虑的解决方案。

---
Learn more:
1. [Reactor模式详解-腾讯云开发者社区-腾讯云](https://cloud.tencent.com/developer/article/2112548)
2. [【Netty】--Reactor模式详解 - aspirant - 博客园](https://www.cnblogs.com/aspirant/p/16149540.html)
3. [Reactor模式详解 - 朱子威 - 博客园](https://www.cnblogs.com/minikobe/p/12160233.html)

详解Reactor模式

## 操作sqlite
```python
'''
Author: yowayimono
Date: 2023-11-18 16:11:04
LastEditors: yowayimono
LastEditTime: 2023-11-18 16:16:11
Description: nothing
'''
import sqlite3
import csv



# 连接数据库
conn = sqlite3.connect('testbook.db')  
cursor = conn.cursor()

# 创建学生信息表
cursor.execute("""CREATE TABLE IF NOT EXISTS student(
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      name TEXT,
      department TEXT, 
      age INTEGER,
      phone TEXT,
      qq TEXT);
      """)

# 读取csv文件插入数据库      
with open('test.csv',encoding='utf8') as f:
  reader = csv.DictReader(f)
  for row in reader:
    name = row['name']
    department = row['department']
    age = row['age']  
    phone = row['phone']
    qq = row['qq']
    
    cursor.execute(
      "INSERT INTO student (name, department, age, phone, qq) VALUES (?,?,?,?,?)",
      (name, department, age, phone, qq))
 
# 查询所有记录   
cursor.execute("SELECT * FROM student")
for line in cursor.fetchall():
  print(line,'\n')

# 修改记录
cursor.execute("UPDATE student SET age=? WHERE name=?", (28, "张三"))

# 删除记录
cursor.execute("DELETE FROM student WHERE name=?", ["李四"])

# 关闭数据库连接
conn.commit()
cursor.close()
conn.close()
```

test.csv文件
```csv
name,department,age,phone,qq
张三,计算机系,20,13000000000,123456 
李四,外语系,21,15000000000,456789
王五,艺术系,22,18000000000,987654
赵六,体育系,23,13000000001,234567
孙齐,数学系,24,15000000001,345678
冯翔,生物系,25,18000000001,456123
陈七,化学系,26,13000000012,567890
于八,物理系,27,15000000012,654321
马九,音乐系,28,18000000012,147852
 吴十,汉语系,29,18500000012,147258
高世芳,历史系,20,18700000012,258369
谢诗梅,铁工系,21,18900000012,369258
王婉如,法律系,22,19100000012,458796
赵昌恺,管理系,23,19300000012,963258
王东方,计算机系,24,19500000012,963258
张萌萍,汉语系,25,19700000012,963258  
张强,设计系,21,19900000012,963258
刘宏伟,体育系,23,13000000013,963258
张超,环境系,20,15000000013,963258
李林,外语系,21,17000000013,963258
黄雨萍,艺术系,22,19000000013,963258
傅岩,物理系,23,21000000013,963258
王军,商务系,24,23000000013,963258
李娜,语言系,25,25000000013,963258
孙颖,心理系,21,27000000013,963258
张云帆,政法系,23,29000000013,963258
王晓雨,教育系,20,310000013,963258
王婷,建筑系,21,330000013,963258
祝颖,美术系,22,350000013,963258
于丽,信息系,23,370000013,963258
刘雪妮,艺术系,24,390000013,963258
李涵,传媒系,25,410000013,963258
赵鹏程,体育系,21,430000013,963258
姜亮,电影系,23,450000013,963258
马文杰,汉语系,20,470000013,963258
邓灵怡,数学系,21,490000013,963258
陈风,艺术系,22,510000013,963258
韩冰,英语系,23,530000013,963258
孙超,计算机系,24,550000013,963258
张玉凤,经济系,25,570000013,963258
刘芳,商务系,21,590000013,963258
陈伟强,管理系,23,610000013,963258
李丽英,植物系,20,630000013,963258
毛毅,汽车系,21,650000013,963258
赵欣然,设计系,22,670000013,963258
肖婷婷,法律系,23,690000013,963258
```
## 生成验证码

```python
import random

red_balls = random.sample(range(1,34), 6)  
red_balls.sort()
blue_ball = random.randint(1,16)

print("红球号码:", red_balls)
print("蓝球号码:", blue_ball)


##################################################################################


import random

num1 = random.randint(0, 9)
num2 = random.randint(0, 9)
num3 = random.randint(0, 9)
num4 = random.randint(0, 9)

captcha = num1 * 1000 + num2 * 100 + num3 * 10 + num4

print("您的验证码是:", captcha)


##################################################################################


import re


with open('func.py') as f:
  text = f.read()
  funcs = re.findall('def ([a-zA-Z_][a-zA-Z0-9_]*)\(', text)
  print(funcs)
  
  
##################################################################################



def func1():
  pass

def funk():
    print('Funk')

def func2():
    pass


```


## 登录界面
```python
import tkinter as tk
from tkinter import ttk

# 登录函数
def login():
  user = entry_user.get()
  pwd = entry_pwd.get()

  if user=="admin" and pwd=="123456":
    label_info["text"] = "登录成功"
  else:  
    label_info["text"] = "用户名或密码错误"
      
# GUI代码
window = tk.Tk()
window.title("登录窗口")

# 标签
label_user = tk.Label(window, text="用户名:") 
label_pwd = tk.Label(window, text="密码:")
label_info = tk.Label(window)

# 入口框
entry_user = tk.Entry(window)
entry_pwd = tk.Entry(window) 

# 按钮
button_login = tk.Button(window, text="登录", command=login)
button_exit = tk.Button(window, text="退出", command=window.quit) 

# 布局
label_user.grid(row=0, column=0)
entry_user.grid(row=0, column=1)
label_pwd.grid(row=1, column=0)  
entry_pwd.grid(row=1, column=1)
button_login.grid(row=2, column=1)
button_exit.grid(row=2, column=0)
label_info.grid(row=3, column=1)

window.mainloop()
```

## 数据分析可视化
```python
'''
Author: yowayimono
Date: 2023-11-18 16:38:21
LastEditors: yowayimono
LastEditTime: 2023-11-18 16:38:25
Description: nothing
'''
import csv
import matplotlib.pyplot as plt

# 读取CSV文件中的数据
data = []
with open('xx.csv', 'r') as csvfile:
    reader = csv.reader(csvfile)
    next(reader)  # 跳过表头
    for row in reader:
        data.append(row)

# 提取季节和水果销售量数据
seasons = [row[0] for row in data]
sales = [int(row[1]) for row in data]

# 绘制折线图
plt.plot(seasons, sales)
plt.xlabel('Season')
plt.ylabel('Sales')
plt.title('Fruit Sales by Season')
plt.show()

# 绘制柱状图
plt.bar(seasons, sales)
plt.xlabel('Season')
plt.ylabel('Sales')
plt.title('Fruit Sales by Season')
plt.show()

# 绘制饼图
plt.pie(sales, labels=seasons, autopct='%1.1f%%')
plt.title('Fruit Sales by Season')
plt.show()

# 绘制散点图
plt.scatter(seasons, sales)
plt.xlabel('Season')
plt.ylabel('Sales')
plt.title('Fruit Sales by Season')
plt.show()


#######################################################################################data

Season,Sales
Spring,120
Summer,180
Autumn,150
Winter,90

Python结课实验

一切还是要从这里考试说起
```c
void InitServerLast(void) {
    bioInit();
    initThreadedIO();
    set_jemalloc_bg_thread(server.jemalloc_bg_thread);
    server.initial_memory_usage = zmalloc_used_memory();
}
```
进入这个函数



```c
/* ==========================================================================
 * Threaded I/O
 * ========================================================================== */

#define IO_THREADS_MAX_NUM 128  //最多128个线程
#ifndef CACHE_LINE_SIZE
#if defined(__aarch64__) && defined(__APPLE__)
#define CACHE_LINE_SIZE 128
#else
#define CACHE_LINE_SIZE 64
#endif
#endif

typedef struct __attribute__((aligned(CACHE_LINE_SIZE))) threads_pending {
    redisAtomic unsigned long value;
} threads_pending;

pthread_t io_threads[IO_THREADS_MAX_NUM];
pthread_mutex_t io_threads_mutex[IO_THREADS_MAX_NUM];
threads_pending io_threads_pending[IO_THREADS_MAX_NUM]; //等待处理的客户端个数
int io_threads_op;      /* IO_THREADS_OP_IDLE, IO_THREADS_OP_READ or IO_THREADS_OP_WRITE. */ // TODO: should access to this be atomic??!



//**************************************************************************
/* Initialize the data structures needed for threaded I/O. */
void initThreadedIO(void) {
    server.io_threads_active = 0; /* We start with threads not active. */

    /* Indicate that io-threads are currently idle */
    io_threads_op = IO_THREADS_OP_IDLE; 

    /* Don't spawn any thread if the user selected a single thread:
     * we'll handle I/O directly from the main thread. */
    if (server.io_threads_num == 1) return; 默认就一个IO线程

    if (server.io_threads_num > IO_THREADS_MAX_NUM) {
        serverLog(LL_WARNING,"Fatal: too many I/O threads configured. "
                             "The maximum number is %d.", IO_THREADS_MAX_NUM);
        exit(1);
    }

    /* Spawn and initialize the I/O threads. */
    for (int i = 0; i < server.io_threads_num; i++) {
        /* Things we do for all the threads including the main thread. */
        io_threads_list[i] = listCreate();
        if (i == 0) continue; /* Thread 0 is the main thread. */

        /* Things we do only for the additional threads. */
        pthread_t tid;
        pthread_mutex_init(&io_threads_mutex[i],NULL);
        setIOPendingCount(i, 0);
        pthread_mutex_lock(&io_threads_mutex[i]); /* Thread will be stopped. */
        if (pthread_create(&tid,NULL,IOThreadMain,(void*)(long)i) != 0) {
            serverLog(LL_WARNING,"Fatal: Can't initialize IO thread.");
            exit(1);
        }
        io_threads[i] = tid;
    }
}

```
- 每个线程一个任务队列，每个线程一把锁

## 线程运行函数
```c
void *IOThreadMain(void *myid) {
    /* The ID is the thread number (from 0 to server.io_threads_num-1), and is
     * used by the thread to just manipulate a single sub-array of clients. */
    long id = (unsigned long)myid;
    char thdname[16];

    snprintf(thdname, sizeof(thdname), "io_thd_%ld", id);
    redis_set_thread_title(thdname);
    redisSetCpuAffinity(server.server_cpulist);
    makeThreadKillable();//设置线程为随时可取消状态

    // 开始取任务执行
    while(1) {
        /* Wait for start */
        for (int j = 0; j < 1000000; j++) {
            if (getIOPendingCount(id) != 0) break;
        }

        /* Give the main thread a chance to stop this thread. */
        if (getIOPendingCount(id) == 0) {
            pthread_mutex_lock(&io_threads_mutex[id]);
            pthread_mutex_unlock(&io_threads_mutex[id]);
            continue;
        }

        serverAssert(getIOPendingCount(id) != 0);

        /* Process: note that the main thread will never touch our list
         * before we drop the pending count to 0. */
        listIter li;
        listNode *ln;
        listRewind(io_threads_list[id],&li);
        while((ln = listNext(&li))) {
            client *c = listNodeValue(ln);
            if (io_threads_op == IO_THREADS_OP_WRITE) {
                writeToClient(c,0);
            } else if (io_threads_op == IO_THREADS_OP_READ) {
                readQueryFromClient(c->conn);
            } else {
                serverPanic("io_threads_op value is unknown");
            }
        }
        listEmpty(io_threads_list[id]);
        setIOPendingCount(id, 0);
    }
}

```
这句是一个简单的线程池结构，上面基本就是消费者逻辑，接下来讲一下生产者逻辑
```c
    list *clients;              /* List of active clients */
    list *clients_to_close;     /* Clients to close asynchronously */
    list *clients_pending_write; /* There is to write or install handler. */
    list *clients_pending_read;  /* Client has pending read socket buffers. */
    list *slaves, *monitors;    /* List of slaves and MONITORs */
```

## 生产者
推迟客户端读操作
```c
void readQueryFromClient(connection *conn) {
    client *c = connGetPrivateData(conn);
    int nread, big_arg = 0;
    size_t qblen, readlen;

    /* Check if we want to read from the client later when exiting from
     * the event loop. This is the case if threaded I/O is enabled. */
    if (postponeClientRead(c)) return;

    /* Update total number of reads on server */
    atomicIncr(server.stat_total_reads_processed, 1);

```

Redis源码阅读 --redis的的多线程IO

[题目](https://leetcode.cn/problems/maximum-sum-with-exactly-k-elements/description/)



大根堆很简单就解决了
```cpp
class Solution {
public:
    int maximizeSum(vector<int>& nums, int k) {
        priority_queue<int>  maxHeap;
        for(int num: nums) maxHeap.push(num);
        int score = 0;
        while(k--) {
            int m = maxHeap.top();
            maxHeap.pop();
            score += m;
            maxHeap.push(m+1);
        }
        return score;

    }
};
```

每日一题11.16

## 判断是否是哨兵模式
```c
    char *exec_name = strrchr(argv[0], '/');
    if (exec_name == NULL) exec_name = argv[0];
    server.sentinel_mode = checkForSentinelMode(argc,argv, exec_name);
    initServerConfig();
    ACLInit(); /* The ACL subsystem must be initialized ASAP because the
                  basic networking code and client creation depends on it. */
    moduleInitModulesSystem();
    connTypeInitialize();
    ***********************************************************、
    
    


/* Returns 1 if there is --sentinel among the arguments or if
 * executable name contains "redis-sentinel". */
int checkForSentinelMode(int argc, char **argv, char *exec_name) {
    if (strstr(exec_name,"redis-sentinel") != NULL) return 1;

    for (int j = 1; j < argc; j++)
        if (!strcmp(argv[j],"--sentinel")) return 1;
    return 0;
}

```
初始化哨兵模式

```c
/* This function overwrites a few normal Redis config default with Sentinel
 * specific defaults. */
void initSentinelConfig(void) {
    server.port = REDIS_SENTINEL_PORT;
    server.protected_mode = 0; /* Sentinel must be exposed. */
}

void freeSentinelLoadQueueEntry(void *item);

/* Perform the Sentinel mode initialization. */
void initSentinel(void) {
    /* Initialize various data structures. */
    sentinel.current_epoch = 0;
    sentinel.masters = dictCreate(&instancesDictType);
    sentinel.tilt = 0;
    sentinel.tilt_start_time = 0;
    sentinel.previous_time = mstime();
    sentinel.running_scripts = 0;
    sentinel.scripts_queue = listCreate();
    sentinel.announce_ip = NULL;
    sentinel.announce_port = 0;
    sentinel.simfailure_flags = SENTINEL_SIMFAILURE_NONE;
    sentinel.deny_scripts_reconfig = SENTINEL_DEFAULT_DENY_SCRIPTS_RECONFIG;
    sentinel.sentinel_auth_pass = NULL;
    sentinel.sentinel_auth_user = NULL;
    sentinel.resolve_hostnames = SENTINEL_DEFAULT_RESOLVE_HOSTNAMES;
    sentinel.announce_hostnames = SENTINEL_DEFAULT_ANNOUNCE_HOSTNAMES;
    memset(sentinel.myid,0,sizeof(sentinel.myid));
    server.sentinel_config = NULL;
}

```


```c
/* Main state. */
struct sentinelState {
    char myid[CONFIG_RUN_ID_SIZE+1]; /* This sentinel ID. */
    uint64_t current_epoch;         /* Current epoch. */
    dict *masters;      /* Dictionary of master sentinelRedisInstances.
                           Key is the instance name, value is the
                           sentinelRedisInstance structure pointer. */
    int tilt;           /* Are we in TILT mode? */
    int running_scripts;    /* Number of scripts in execution right now. */
    mstime_t tilt_start_time;       /* When TITL started. */
    mstime_t previous_time;         /* Last time we ran the time handler. */
    list *scripts_queue;            /* Queue of user scripts to execute. */
    char *announce_ip;  /* IP addr that is gossiped to other sentinels if
                           not NULL. */
    int announce_port;  /* Port that is gossiped to other sentinels if
                           non zero. */
    unsigned long simfailure_flags; /* Failures simulation. */
    int deny_scripts_reconfig; /* Allow SENTINEL SET ... to change script
                                  paths at runtime? */
    char *sentinel_auth_pass;    /* Password to use for AUTH against other sentinel */
    char *sentinel_auth_user;    /* Username for ACLs AUTH against other sentinel. */
    int resolve_hostnames;       /* Support use of hostnames, assuming DNS is well configured. */
    int announce_hostnames;      /* Announce hostnames instead of IPs when we have them. */
} sentinel;
```

## 解析sentinel.conf
loadServerConfig