简单的服务器搭建

1、资源服务器
可以用python的http.server模块快速搭建一个简单的服务器
方法1、在需要的目录下执行下面命令就可以下载该目录下的资源了。
python3 -m http.server 8000
方法2、创建python脚本，运行脚本，然后在浏览器中输入http://localhost:8000/就可以访问了。

import http.server
import socketserver

PORT = 8000

Handler = http.server.SimpleHTTPRequestHandler

with socketserver.TCPServer(("", PORT), Handler) as httpd:
    print("serving at port", PORT)
    httpd.serve_forever()

2、短连接服务器，一般是web/http 服务器，可以用python 的flask 模块快速搭建一个简单的服务器

from flask import Flask, request, jsonify
from flask_cors import CORS
import sqlite3
import uuid
import threading
import time

app = Flask(__name__)
CORS(app)  # 允许跨域

# 初始化数据库
def init_db():
    conn = sqlite3.connect('game.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS players
                 (id TEXT PRIMARY KEY, 
                  device_id TEXT UNIQUE,
                  progress INTEGER DEFAULT 0)''')
    conn.commit()
    conn.close()

# 匿名注册
@app.route('/register', methods=['POST'])
def register():
    device_id = str(uuid.uuid4())  # 生成设备唯一ID
    conn = sqlite3.connect('game.db')
    c = conn.cursor()
    c.execute("INSERT INTO players (id, device_id) VALUES (?, ?)", 
              (str(uuid.uuid4()), device_id))
    conn.commit()
    conn.close()
    print("[Server] Simulating push notification..." + device_id)
    return jsonify({"code": 200, "device_id": device_id})

# 登录/获取进度
@app.route('/login', methods=['POST'])
def login():
    device_id = request.json.get('device_id')
    conn = sqlite3.connect('game.db')
    c = conn.cursor()
    c.execute("SELECT progress FROM players WHERE device_id=?", (device_id,))
    result = c.fetchone()
    conn.close()
    if result:
        return jsonify({"code": 200, "progress": result[0]})
    else:
        return jsonify({"code": 404, "error": "User not found"}), 404

# 同步进度
@app.route('/sync', methods=['POST'])
def sync():
    device_id = request.json.get('device_id')
    progress = request.json.get('progress')
    conn = sqlite3.connect('game.db')
    c = conn.cursor()
    c.execute("UPDATE players SET progress=? WHERE device_id=?", 
              (progress, device_id))
    conn.commit()
    conn.close()
    return jsonify({"code": 200})

# 模拟服务器推送
def push_notification():
    while True:
        time.sleep(10)  # 每10秒推送一次
        print("[Server] Simulating push notification...")

if __name__ == '__main__':
    init_db()
    # 启动推送线程
    threading.Thread(target=push_notification, daemon=True).start()
    app.run(host='0.0.0.0', port=5000, debug=True)

3、长连接服务器，一般是socket 服务器，可以用python 的socket 模块快速搭建一个简单的服务器

import socket
import sqlite3
import uuid
import json
import threading
from threading import Lock
import time

# 线程安全的客户端连接管理
clients = {}
clients_lock = Lock()

# 初始化数据库
def init_db():
    conn_db = sqlite3.connect('game.db')
    c = conn_db.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS players
                 (id TEXT PRIMARY KEY, 
                  device_id TEXT UNIQUE,
                  progress INTEGER DEFAULT 0)''')
    conn_db.commit()
    conn_db.close()

# 处理客户端连接
def handle_client(conn, addr):
    device_id = None
    try:
        while True:
            # 接收消息长度头 (4字节)
            header = conn.recv(4)
            print(f"收到header: {header}")
            if not header: break
            msg_len = int.from_bytes(header, byteorder='big')
            
            # 接收实际数据
            data = conn.recv(msg_len)
            print(f"收到data: {data}")
            if not data: break
            
            # 解析JSON
            try:
                req = json.loads(data.decode('utf-8'))
                cmd = req.get("cmd")
                print(f"收到cmd: {cmd}")
                # 注册/登录
                if cmd == "register_or_login":
                    device_id = req.get("device_id")
                    if not device_id:
                        device_id = str(uuid.uuid4())
                        conn_db = sqlite3.connect('game.db')
                        c = conn_db.cursor()
                        c.execute("INSERT INTO players (id, device_id) VALUES (?, ?)", 
                                  (str(uuid.uuid4()), device_id))
                        conn_db.commit()
                        conn_db.close()
                    
                    clients[device_id] = conn
                    # 查询进度
                    conn_db = sqlite3.connect('game.db')
                    c = conn_db.cursor()
                    c.execute("SELECT progress FROM players WHERE device_id=?", (device_id,))
                    progress = c.fetchone()[0]
                    conn_db.close()
                    
                    # 返回响应
                    resp = {
                        "cmd": "login_response",
                        "device_id": device_id,
                        "progress": progress
                    }
                    send_response(conn, resp)
                
                # 同步进度
                elif cmd == "sync_progress":
                    progress = req.get("progress")
                    conn_db = sqlite3.connect('game.db')
                    c = conn_db.cursor()
                    c.execute("UPDATE players SET progress=? WHERE device_id=?", 
                              (progress, device_id))
                    conn_db.commit()
                    conn_db.close()
                    
                    send_response(conn,  {
                        "cmd": "sync_ok",
                        "device_id": device_id,
                        "progress": progress
                    })
            
            except json.JSONDecodeError:
                print(f"非法JSON数据: {data}")
    
    except ConnectionResetError:
        print(f"客户端断开: {addr}")
    finally:
        with clients_lock:
            if device_id in clients:
                del clients[device_id]
        conn.close()

# 发送响应（添加长度头）
def send_response(conn, data):
    json_str = json.dumps(data)
    encoded = json_str.encode('utf-8')
    header = len(encoded).to_bytes(4, byteorder='little')
    conn.send(header + encoded)
    print(f"回应cmd: {data.get('cmd')}，data={data}")

# 模拟服务器推送
def push_notification():
    while True:
        time.sleep(10)  # 每10秒推送一次
        message = {
            "cmd": "server_push",
            "content": "新活动已上线！"
        }
        with clients_lock:
            for device_id, conn in list(clients.items()):
                try:
                    send_response(conn, message)
                except:
                    del clients[device_id]

if __name__ == '__main__':
    init_db()
    host, port = "0.0.0.0", 8766
    
    # 启动Socket服务器
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen()
        print(f"Socket服务器启动: {host}:{port}")
        
        threading.Thread(target=push_notification, daemon=True).start()
        while True:
            conn, addr = s.accept()
            threading.Thread(target=handle_client, args=(conn, addr)).start()
            # 启动推送线程

或者用云风大大的 skynet

main.lua
local skynet = require "skynet"

skynet.start(function()
	skynet.error("Server start")
	if not skynet.getenv "daemon" then
		local console = skynet.newservice("console")
	end
	skynet.newservice("debug_console",8000)
	local proto = skynet.uniqueservice "protoloader"
	skynet.call(proto, "lua", "load", {
		"proto.c2s",
		"proto.s2c",
	})
	local hub = skynet.uniqueservice "hub"
	skynet.call(hub, "lua", "open", "0.0.0.0", 5678)
	skynet.exit()
end)
hub.lua
local skynet = require "skynet"
local socket = require "socket"
local proxy = require "socket_proxy"
local log = require "log"
local service = require "service"

local hub = {}
local data = { socket = {} }

local function auth_socket(fd)
	return (skynet.call(service.auth, "lua", "shakehand" , fd))
end

local function assign_agent(fd, userid)
	skynet.call(service.manager, "lua", "assign", fd, userid)
end

function new_socket(fd, addr)
	data.socket[fd] = "[AUTH]"
	proxy.subscribe(fd)
	local ok , userid =  pcall(auth_socket, fd)
	if ok then
		data.socket[fd] = userid
		if pcall(assign_agent, fd, userid) then
			return	-- succ
		else
			log("Assign failed %s to %s", addr, userid)
		end
	else
		log("Auth faild %s", addr)
	end
	proxy.close(fd)
	data.socket[fd] = nil
end

function hub.open(ip, port)
	log("Listen %s:%d", ip, port)
	assert(data.fd == nil, "Already open")
	data.fd = socket.listen(ip, port)
	data.ip = ip
	data.port = port
	socket.start(data.fd, new_socket)
end

function hub.close()
	assert(data.fd)
	log("Close %s:%d", data.ip, data.port)
	socket.close(data.fd)
	data.ip = nil
	data.port = nil
end

service.init {
	command = hub,
	info = data,
	require = {
		"auth",
		"manager",
	}
}
auth.lua
local skynet = require "skynet"
local service = require "service"
local client = require "client"
local log = require "log"

local auth = {}
local users = {}
local cli = client.handler()

local SUCC = { ok = true }
local FAIL = { ok = false }

function cli:signup(args)
	log("signup userid = %s", args.userid)
	if users[args.userid] then
		return FAIL
	else
		users[args.userid] = true
		return SUCC
	end
end

function cli:signin(args)
	log("signin userid = %s", args.userid)
	if users[args.userid] then
		self.userid = args.userid
		self.exit = true
		return SUCC
	else
		return FAIL
	end
end

function cli:ping()
	log("ping")
end

function auth.shakehand(fd)
	local c = client.dispatch { fd = fd }
	return c.userid
end

service.init {
	command = auth,
	info = users,
	init = client.init "proto",
}
manager.lua
local skynet = require "skynet"
local service = require "service"
local log = require "log"

local manager = {}
local users = {}

local function new_agent()
	-- todo: use a pool
	return skynet.newservice "agent"
end

local function free_agent(agent)
	-- kill agent, todo: put it into a pool maybe better
	skynet.kill(agent)
end

function manager.assign(fd, userid)
	local agent
	repeat
		agent = users[userid]
		if not agent then
			agent = new_agent()
			if not users[userid] then
				-- double check
				users[userid] = agent
			else
				free_agent(agent)
				agent = users[userid]
			end
		end
	until skynet.call(agent, "lua", "assign", fd, userid)
	log("Assign %d to %s [%s]", fd, userid, agent)
end

function manager.exit(userid)
	users[userid] = nil
end

service.init {
	command = manager,
	info = users,
}
agent.lua
local skynet = require "skynet"
local service = require "service"
local client = require "client"
local log = require "log"

local agent = {}
local data = {}
local cli = client.handler()

function cli:ping()
	assert(self.login)
	log "ping"
end

function cli:login()
	assert(not self.login)
	if data.fd then
		log("login fail %s fd=%d", data.userid, self.fd)
		return { ok = false }
	end
	data.fd = self.fd
	self.login = true
	log("login succ %s fd=%d", data.userid, self.fd)
	client.push(self, "push", { text = "welcome" })	-- push message to client
	return { ok = true }
end

local function new_user(fd)
	local ok, error = pcall(client.dispatch , { fd = fd })
	log("fd=%d is gone. error = %s", fd, error)
	client.close(fd)
	if data.fd == fd then
		data.fd = nil
		skynet.sleep(1000)	-- exit after 10s
		if data.fd == nil then
			-- double check
			if not data.exit then
				data.exit = true	-- mark exit
				skynet.call(service.manager, "lua", "exit", data.userid)	-- report exit
				log("user %s afk", data.userid)
				skynet.exit()
			end
		end
	end
end

function agent.assign(fd, userid)
	if data.exit then
		return false
	end
	if data.userid == nil then
		data.userid = userid
	end
	assert(data.userid == userid)
	skynet.fork(new_user, fd)
	return true
end

service.init {
	command = agent,
	info = data,
	require = {
		"manager",
	},
	init = client.init "proto",
}
protoloader.lua
local skynet = require "skynet"
local sprotoparser = require "sprotoparser"
local sprotoloader = require "sprotoloader"
local service = require "service"
local log = require "log"

local loader = {}
local data = {}

local function load(name)
	local filename = string.format("proto/%s.sproto", name)
	local f = assert(io.open(filename), "Can't open " .. name)
	local t = f:read "a"
	f:close()
	return sprotoparser.parse(t)
end

function loader.load(list)
	for i, name in ipairs(list) do
		local p = load(name)
		log("load proto [%s] in slot %d", name, i)
		data[name] = i
		sprotoloader.save(p, i)
	end
end

function loader.index(name)
	return data[name]
end

service.init {
	command = loader,
	info = data
}