跳到主要内容

· 阅读需 1 分钟
forkp

使用Python和Golang测试10000个连接​

python

import asyncio
import websockets

async def connect(uri):
    async with websockets.connect(uri) as websocket:
        # 在这里可以发送和接收消息
        await websocket.send("Hello")
        response = await websocket.recv()
        print(response)

async def main():
    uri = "ws://localhost:3322"
    tasks = [connect(uri) for _ in range(10000)]
    await asyncio.gather(*tasks)

asyncio.run(main())

go

package main

import (
	"log"
	"sync"
	"time"
	"golang.org/x/net/websocket"
)

const numConnections = 10000

func connect(uri string, wg *sync.WaitGroup) {
	defer wg.Done()
	ws, err := websocket.Dial(uri, "", "http://localhost/")
	if err != nil {
		log.Println("Failed to connect:", err)
		return
	}
	defer ws.Close()

	// 发送和接收消息
	_, err = ws.Write([]byte("Hello"))
	if err != nil {
		log.Println("Failed to send message:", err)
		return
	}

	var msg = make([]byte, 512)
	_, err = ws.Read(msg)
	if err != nil {
		log.Println("Failed to read message:", err)
		return
	}
	log.Println("Received message:", string(msg))
}

func main() {
	uri := "ws://localhost:3322"
	var wg sync.WaitGroup

	for i := 0; i < numConnections; i++ {
		wg.Add(1)
		go connect(uri, &wg)
		time.Sleep(time.Millisecond) // 控制连接创建速率
	}

	wg.Wait()
}

· 阅读需 1 分钟
forkp

开启透明支持​

在开启透明支持后，需要设置网页的背景为透明色，不然开启不了。


html,
body
{
    background-color: transparent;
}

· 阅读需 1 分钟
forkp

Python 代码如下​

import asyncio
import aiohttp
import json
import redis
from urllib.parse import urlparse, parse_qs

s = {}

def cache(db=0):
    return redis.Redis(host='127.0.0.1', port=6379, decode_responses=True,db=db)

async def fetch(session, url):
    async with session.get(url,ssl=False) as response:
        parsed_url = urlparse(url)
        query_params = parse_qs(parsed_url.query)
        from_money_value = query_params.get('from_money', [''])[0]
        k = await response.text()
        kd = json.loads(k)
        money_value = {"money1":find_key_value(kd, 'money1') or [],"money2":find_key_value(kd, 'money2') or [],"money2_num": find_key_value(kd, 'money2_num') or [],"money2_rev":find_key_value(kd, 'money2_rev') or []}
        s[from_money_value] = money_value
        money_value = json.dumps(money_value)
        k_s = "laravel_database_money:{}".format(from_money_value)
        cache(db=1).set(k_s,money_value)
        return k

async def main():
    url_cny_usd = 'https://sp0.baidu.com/5LMDcjW6BwF3otqbppnN2DJv/finance.pae.baidu.com/vapi/async/v1?from_money=CNY&to_money=USD&srcid=5293'
    async with aiohttp.ClientSession() as session:
        cny_usd_response = await fetch(session, url_cny_usd)
        data = json.loads(cny_usd_response)
        money_list = find_key_value(data, 'money_list') or []
        currency_codes = [item[1].split(" ")[1] for item in money_list if isinstance(item, list) and len(item) > 0]
        urls = [f'https://sp0.baidu.com/5LMDcjW6BwF3otqbppnN2DJv/finance.pae.baidu.com/vapi/async/v1?from_money={code}&to_money=USD&srcid=5293' for code in currency_codes]
        
        tasks = [fetch(session, url) for url in urls]
        responses = await asyncio.gather(*tasks)
        k_s = "laravel_database_money_all"
        cache(db=1).set(k_s,json.dumps(s))
        
def find_key_value(nested_item, key):
    if isinstance(nested_item, dict):
        if key in nested_item:
            return nested_item[key]  
        for v in nested_item.values():
            found_value = find_key_value(v, key)
            if found_value is not None:
                return found_value
    elif isinstance(nested_item, list):
        for item in nested_item:
            found_value = find_key_value(item, key)
            if found_value is not None:
                return found_value
    return None 


asyncio.run(main())

· 阅读需 1 分钟
forkp

css 代码​

*::-webkit-scrollbar {
  width: 8px;
}

*::-webkit-scrollbar-thumb {
  border-radius: 10px;
  box-shadow: inset 0 0 5px rgba(236, 236, 236, 0.1);
  background: #adacac;
}

*::-webkit-scrollbar-track {
  box-shadow: inset 0 0 5px rgba(236, 236, 236, 0.8);
  border-radius: 10px;
  background: #ededed;
}

· 阅读需 1 分钟
forkp

前言​

使用 pip install redis的时候提示错误

WARNING: pip is configured with locations that require TLS/SSL, however the ssl module in Python is not available.

由于centos7的套件比较老旧，所以得升级openssl库

执行命令安装

sudo yum install openssl11 openssl11-devel
mkdir /usr/local/openssl11
cd /usr/local/openssl11
ln -s /usr/lib64/openssl11 lib
ln -s /usr/include/openssl11 include

编译 Python3.12​

./configure --prefix=/usr/local/python3.12 --with-ssl=/usr/local/openssl11
make && make install

搞定收工