WebSocket：实现实时通信

go get github.com/gorilla/websocket

package main

import (
	"log"
	"net/http"

	"github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true // 允许所有来源（生产环境应该限制）
	},
}

func echoHandler(w http.ResponseWriter, r *http.Request) {
	// 升级 HTTP 连接为 WebSocket
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("升级失败:", err)
		return
	}
	defer conn.Close()

	log.Println("新连接:", conn.RemoteAddr())

	// 循环读取和发送消息
	for {
		messageType, message, err := conn.ReadMessage()
		if err != nil {
			log.Println("读取失败:", err)
			break
		}

		log.Printf("收到: %s", message)

		// Echo 回去
		err = conn.WriteMessage(messageType, message)
		if err != nil {
			log.Println("写入失败:", err)
			break
		}
	}
}

func main() {
	http.HandleFunc("/echo", echoHandler)

	log.Println("WebSocket 服务器启动在 :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
	"bufio"
	"log"
	"os"

	"github.com/gorilla/websocket"
)

func main() {
	// 连接 WebSocket 服务器
	url := "ws://localhost:8080/echo"
	conn, _, err := websocket.DefaultDialer.Dial(url, nil)
	if err != nil {
		log.Fatal("连接失败:", err)
	}
	defer conn.Close()

	// 启动 goroutine 接收消息
	go func() {
		for {
			_, message, err := conn.ReadMessage()
			if err != nil {
				log.Println("读取失败:", err)
				return
			}
			log.Printf("收到: %s", message)
		}
	}()

	// 从标准输入读取消息并发送
	scanner := bufio.NewScanner(os.Stdin)
	log.Println("输入消息（按 Enter 发送）:")
	for scanner.Scan() {
		message := scanner.Text()
		err := conn.WriteMessage(websocket.TextMessage, []byte(message))
		if err != nil {
			log.Println("发送失败:", err)
			break
		}
	}
}

package main

import (
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

const (
	writeWait      = 10 * time.Second
	pongWait       = 60 * time.Second
	pingPeriod     = (pongWait * 9) / 10
	maxMessageSize = 512
)

type Client struct {
	conn     *websocket.Conn
	send     chan []byte
	room     *ChatRoom
}

type ChatRoom struct {
	clients    map[*Client]bool
	broadcast  chan []byte
	register   chan *Client
	unregister chan *Client
	mu         sync.RWMutex
}

func NewChatRoom() *ChatRoom {
	return &ChatRoom{
		clients:    make(map[*Client]bool),
		broadcast:  make(chan []byte),
		register:   make(chan *Client),
		unregister: make(chan *Client),
	}
}

func (room *ChatRoom) Run() {
	for {
		select {
		case client := <-room.register:
			room.mu.Lock()
			room.clients[client] = true
			room.mu.Unlock()
			log.Printf("新客户端连接，当前在线: %d", len(room.clients))
		case client := <-room.unregister:
			room.mu.Lock()
			if _, ok := room.clients[client]; ok {
				delete(room.clients, client)
				close(client.send)
				log.Printf("客户端断开，当前在线: %d", len(room.clients))
			}
			room.mu.Unlock()
		case message := <-room.broadcast:
			room.mu.RLock()
			for client := range room.clients {
				select {
				case client.send <- message:
				default:
					close(client.send)
					delete(room.clients, client)
				}
			}
			room.mu.RUnlock()
		}
	}
}

func (c *Client) readPump() {
	defer func() {
		c.room.unregister <- c
		c.conn.Close()
	}()

	c.conn.SetReadLimit(maxMessageSize)
	c.conn.SetReadDeadline(time.Now().Add(pongWait))
	c.conn.SetPongHandler(func(string) error {
		c.conn.SetReadDeadline(time.Now().Add(pongWait))
		return nil
	})

	for {
		_, message, err := c.conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("错误: %v", err)
			}
			break
		}
		c.room.broadcast <- message
	}
}

func (c *Client) writePump() {
	ticker := time.NewTicker(pingPeriod)
	defer func() {
		ticker.Stop()
		c.conn.Close()
	}()

	for {
		select {
		case message, ok := <-c.send:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}

			w, err := c.conn.NextWriter(websocket.TextMessage)
			if err != nil {
				return
			}
			w.Write(message)

			if err := w.Close(); err != nil {
				return
			}

		case <-ticker.C:
			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

func serveWs(room *ChatRoom, w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println(err)
		return
	}
	client := &Client{conn: conn, send: make(chan []byte, 256), room: room}
	client.room.register <- client

	go client.writePump()
	go client.readPump()
}

func main() {
	room := NewChatRoom()
	go room.Run()

	http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
		serveWs(room, w, r)
	})

	log.Println("WebSocket 服务器启动在 :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
	"log"
	"net/http"
	"sync"

	"github.com/gorilla/websocket"
)

// Client 表示一个聊天室用户
type Client struct {
	conn     *websocket.Conn
	username string
	send     chan []byte
}

// ChatRoom 管理所有连接
type ChatRoom struct {
	clients    map[*Client]bool
	broadcast  chan []byte
	register   chan *Client
	unregister chan *Client
	mu         sync.RWMutex
}

func NewChatRoom() *ChatRoom {
	return &ChatRoom{
		clients:    make(map[*Client]bool),
		broadcast:  make(chan []byte),
		register:   make(chan *Client),
		unregister: make(chan *Client),
	}
}

func (room *ChatRoom) Run() {
	for {
		select {
		case client := <-room.register:
			room.mu.Lock()
			room.clients[client] = true
			room.mu.Unlock()
			log.Printf("用户 %s 加入聊天室", client.username)

			// 广播加入消息
			room.broadcast <- []byte("系统: " + client.username + " 加入了聊天室")

		case client := <-room.unregister:
			room.mu.Lock()
			if _, ok := room.clients[client]; ok {
				delete(room.clients, client)
				close(client.send)
				log.Printf("用户 %s 离开聊天室", client.username)
				room.broadcast <- []byte("系统: " + client.username + " 离开了聊天室")
			}
			room.mu.Unlock()

		case message := <-room.broadcast:
			room.mu.RLock()
			for client := range room.clients {
				select {
				case client.send <- message:
				default:
					close(client.send)
					delete(room.clients, client)
				}
			}
			room.mu.RUnlock()
		}
	}
}

func (room *ChatRoom) handleClient(conn *websocket.Conn, username string) {
	client := &Client{
		conn:     conn,
		username: username,
		send:     make(chan []byte, 256),
	}

	room.register <- client

	// 启动写 goroutine
	go func() {
		defer conn.Close()
		for message := range client.send {
			err := conn.WriteMessage(websocket.TextMessage, message)
			if err != nil {
				break
			}
		}
	}()

	// 读消息
	defer func() {
		room.unregister <- client
		conn.Close()
	}()

	for {
		_, message, err := conn.ReadMessage()
		if err != nil {
			break
		}

		// 格式化消息
		formatted := []byte(username + ": " + string(message))
		room.broadcast <- formatted
	}
}

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

var room = NewChatRoom()

func chatHandler(w http.ResponseWriter, r *http.Request) {
	username := r.URL.Query().Get("username")
	if username == "" {
		username = "匿名用户"
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("升级失败:", err)
		return
	}

	room.handleClient(conn, username)
}

func main() {
	go room.Run()

	http.HandleFunc("/chat", chatHandler)

	log.Println("聊天室服务器启动在 :8080")
	log.Println("访问: ws://localhost:8080/chat?username=你的名字")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

type Notification struct {
	UserID  string    `json:"user_id"`
	Title   string    `json:"title"`
	Body    string    `json:"body"`
	Type    string    `json:"type"`
	Time    time.Time `json:"time"`
}

type NotificationHub struct {
	clients map[string]*websocket.Conn // user_id -> conn
	mu      sync.RWMutex
	notifCh chan Notification
}

func NewNotificationHub() *NotificationHub {
	return &NotificationHub{
		clients: make(map[string]*websocket.Conn),
		notifCh: make(chan Notification, 1000),
	}
}

func (h *NotificationHub) Register(userID string, conn *websocket.Conn) {
	h.mu.Lock()
	if oldConn, ok := h.clients[userID]; ok {
		oldConn.Close()
	}
	h.clients[userID] = conn
	h.mu.Unlock()
	log.Printf("用户 %s 已连接通知服务", userID)
}

func (h *NotificationHub) Unregister(userID string) {
	h.mu.Lock()
	if conn, ok := h.clients[userID]; ok {
		conn.Close()
		delete(h.clients, userID)
		log.Printf("用户 %s 已断开通知服务", userID)
	}
	h.mu.Unlock()
}

func (h *NotificationHub) DispatchLoop() {
	for notif := range h.notifCh {
		h.mu.RLock()
		conn, ok := h.clients[notif.UserID]
		h.mu.RUnlock()
		if !ok {
			log.Printf("用户 %s 不在线，通知暂存或丢弃", notif.UserID)
			continue
		}
		data, _ := json.Marshal(notif)
		if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
			log.Printf("向用户 %s 推送通知失败: %v", notif.UserID, err)
			h.Unregister(notif.UserID)
		}
	}
}

func (h *NotificationHub) Push(notif Notification) {
	select {
	case h.notifCh <- notif:
	default:
		log.Println("通知队列已满，丢弃消息")
	}
}

var (
	upgrader        = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}
	notificationHub = NewNotificationHub()
)

func notifyHandler(w http.ResponseWriter, r *http.Request) {
	userID := r.URL.Query().Get("user_id")
	if userID == "" {
		http.Error(w, "user_id is required", http.StatusBadRequest)
		return
	}
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("升级失败:", err)
		return
	}
	notificationHub.Register(userID, conn)
	defer notificationHub.Unregister(userID)

	for {
		_, _, err := conn.ReadMessage()
		if err != nil {
			log.Printf("用户 %s 连接异常: %v", userID, err)
			break
		}
	}
}

func triggerHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "POST only", http.StatusMethodNotAllowed)
		return
	}
	var notif Notification
	if err := json.NewDecoder(r.Body).Decode(&notif); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	notif.Time = time.Now()
	notificationHub.Push(notif)
	w.Write([]byte("通知已推送"))
}

func main() {
	go notificationHub.DispatchLoop()

	http.HandleFunc("/notify", notifyHandler)
	http.HandleFunc("/trigger", triggerHandler)

	log.Println("通知服务器启动在 :8080")
	log.Println("连接: ws://localhost:8080/notify?user_id=xxx")
	log.Println("推送通知: POST /trigger {\"user_id\":\"xxx\",\"title\":\"...\",\"body\":\"...\",\"type\":\"...\"}")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

type Player struct {
	ID       string  `json:"id"`
	Nickname string  `json:"nickname"`
	X        float64 `json:"x"`
	Y        float64 `json:"y"`
	Conn     *websocket.Conn
}

type PlayerEvent struct {
	Type     string  `json:"type"`
	PlayerID string  `json:"player_id"`
	X        float64 `json:"x,omitempty"`
	Y        float64 `json:"y,omitempty"`
}

type GameRoom struct {
	ID        string
	players   map[string]*Player
	mu        sync.RWMutex
	broadcast chan []byte
	ticker    *time.Ticker
	done      chan struct{}
}

func NewGameRoom(id string) *GameRoom {
	return &GameRoom{
		ID:        id,
		players:   make(map[string]*Player),
		broadcast: make(chan []byte, 100),
		done:      make(chan struct{}),
	}
}

func (r *GameRoom) Join(player *Player) {
	r.mu.Lock()
	r.players[player.ID] = player
	r.mu.Unlock()

	// 通知房间内其他玩家有新玩家加入
	msg, _ := json.Marshal(PlayerEvent{Type: "player_joined", PlayerID: player.ID})
	r.broadcast <- msg
	log.Printf("玩家 %s 加入房间 %s，当前人数: %d", player.Nickname, r.ID, len(r.players))
}

func (r *GameRoom) Leave(playerID string) {
	r.mu.Lock()
	if player, ok := r.players[playerID]; ok {
		player.Conn.Close()
		delete(r.players, playerID)
	}
	r.mu.Unlock()

	msg, _ := json.Marshal(PlayerEvent{Type: "player_left", PlayerID: playerID})
	r.broadcast <- msg
	log.Printf("玩家 %s 离开房间 %s，当前人数: %d", playerID, r.ID, len(r.players))
}

func (r *GameRoom) UpdatePosition(playerID string, x, y float64) {
	r.mu.Lock()
	if player, ok := r.players[playerID]; ok {
		player.X = x
		player.Y = y
	}
	r.mu.Unlock()

	msg, _ := json.Marshal(PlayerEvent{Type: "move", PlayerID: playerID, X: x, Y: y})
	r.broadcast <- msg
}

func (r *GameRoom) Run() {
	r.ticker = time.NewTicker(100 * time.Millisecond)
	defer r.ticker.Stop()

	for {
		select {
		case msg := <-r.broadcast:
			r.mu.RLock()
			for _, player := range r.players {
				if err := player.Conn.WriteMessage(websocket.TextMessage, msg); err != nil {
					log.Printf("向玩家 %s 发送消息失败", player.ID)
				}
			}
			r.mu.RUnlock()
		case <-r.ticker.C:
			// 可以在这里实现定时广播完整游戏状态
		case <-r.done:
			return
		}
	}
}

type GameServer struct {
	rooms map[string]*GameRoom
	mu    sync.RWMutex
}

func NewGameServer() *GameServer {
	return &GameServer{rooms: make(map[string]*GameRoom)}
}

func (s *GameServer) GetOrCreateRoom(roomID string) *GameRoom {
	s.mu.Lock()
	defer s.mu.Unlock()
	if room, ok := s.rooms[roomID]; ok {
		return room
	}
	room := NewGameRoom(roomID)
	s.rooms[roomID] = room
	go room.Run()
	return room
}

var (
	upgrader   = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}
	gameServer = NewGameServer()
)

func gameHandler(w http.ResponseWriter, r *http.Request) {
	roomID := r.URL.Query().Get("room")
	playerID := r.URL.Query().Get("player_id")
	nickname := r.URL.Query().Get("nickname")
	if roomID == "" || playerID == "" {
		http.Error(w, "room and player_id required", http.StatusBadRequest)
		return
	}
	if nickname == "" {
		nickname = playerID
	}

	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("升级失败:", err)
		return
	}

	room := gameServer.GetOrCreateRoom(roomID)
	player := &Player{ID: playerID, Nickname: nickname, X: 0, Y: 0, Conn: conn}
	room.Join(player)
	defer room.Leave(playerID)

	for {
		_, data, err := conn.ReadMessage()
		if err != nil {
			break
		}
		var event PlayerEvent
		if err := json.Unmarshal(data, &event); err != nil {
			continue
		}
		if event.Type == "move" {
			room.UpdatePosition(playerID, event.X, event.Y)
		}
	}
}

func main() {
	http.HandleFunc("/game", gameHandler)
	log.Println("游戏服务器启动在 :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
	"encoding/json"
	"log"
	"math/rand"
	"net/http"
	"time"

	"github.com/gorilla/websocket"
)

type Stock struct {
	Symbol string  `json:"symbol"`
	Price  float64 `json:"price"`
	Change float64 `json:"change"`
	Time   string  `json:"time"`
}

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool {
		return true
	},
}

func stockHandler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("升级失败:", err)
		return
	}
	defer conn.Close()

	// 模拟股票数据
	stocks := []Stock{
		{Symbol: "AAPL", Price: 150.00},
		{Symbol: "GOOGL", Price: 2800.00},
		{Symbol: "MSFT", Price: 300.00},
		{Symbol: "TSLA", Price: 700.00},
	}

	// 定期推送行情
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			// 随机更新价格
			for i := range stocks {
				change := (rand.Float64() - 0.5) * 10
				stocks[i].Price += change
				stocks[i].Change = change
				stocks[i].Time = time.Now().Format("15:04:05")
			}

			// 发送数据
			data, _ := json.Marshal(stocks)
			err := conn.WriteMessage(websocket.TextMessage, data)
			if err != nil {
				log.Println("写入失败:", err)
				return
			}
		}
	}
}

func main() {
	http.HandleFunc("/stocks", stockHandler)

	// 提供 HTML 页面
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/html")
		w.Write([]byte(`
<!DOCTYPE html>
<html>
<head>
    <title>实时股票行情</title>
    <style>
        body { font-family: Arial, sans-serif; padding: 20px; }
        .stock { margin: 10px 0; padding: 10px; border: 1px solid #ccc; }
        .up { color: green; }
        .down { color: red; }
    </style>
</head>
<body>
    <h1>实时股票行情</h1>
    <div id="stocks"></div>

    <script>
        const ws = new WebSocket('ws://localhost:8080/stocks');

        ws.onmessage = function(event) {
            const stocks = JSON.parse(event.data);
            const container = document.getElementById('stocks');

            container.innerHTML = stocks.map(stock => `
                <div class="stock">
                    <strong>${stock.symbol}</strong>:
                    $${stock.price.toFixed(2)}
                    <span class="${stock.change >= 0 ? 'up' : 'down'}">
                        ${stock.change >= 0 ? '↑' : '↓'} ${Math.abs(stock.change).toFixed(2)}
                    </span>
                    <small>(${stock.time})</small>
                </div>
            `).join('');
        };

        ws.onclose = function() {
            document.getElementById('stocks').innerHTML = '连接已断开';
        };
    </script>
</body>
</html>
		`))
	})

	log.Println("股票行情服务器启动在 :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}

package main

import (
	"log"
	"net/http"
	"sync/atomic"

	"github.com/gorilla/websocket"
)

var activeConnections int64

var upgrader = websocket.Upgrader{
	CheckOrigin: func(r *http.Request) bool { return true },
}

func connCountHandler(w http.ResponseWriter, r *http.Request) {
	conn, err := upgrader.Upgrade(w, r, nil)
	if err != nil {
		log.Println("升级失败:", err)
		return
	}
	atomic.AddInt64(&activeConnections, 1)
	log.Printf("新连接建立，当前活跃连接数: %d", atomic.LoadInt64(&activeConnections))
	defer func() {
		conn.Close()
		atomic.AddInt64(&activeConnections, -1)
		log.Printf("连接断开，当前活跃连接数: %d", atomic.LoadInt64(&activeConnections))
	}()

	for {
		_, _, err := conn.ReadMessage()
		if err != nil {
			break
		}
	}
}

func main() {
	http.HandleFunc("/ws", connCountHandler)
	log.Println("服务器启动在 :8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}