Golang搭建即時通信系統IM-System

package main

import (
	"fmt"
	"io"
	"net"
	"sync"
	"time"
)

type Server struct {
	Ip   string
	Port int

	//用戶在線串列
	OnlineMap map[string]*User
	//給全域變數map加同步鎖
	mapLock sync.RWMutex

	// 訊息廣播的channel
	Message chan string
}

//創建一個普通的server介面（函式）
//開頭大寫表示對外開放
func NewServer(ip string, port int) *Server {
	server := &Server{
		Ip: ip,
		Port: port,
		OnlineMap: make(map[string]*User),
		Message: make(chan string),
	}
	return server
}

//監聽Message廣播訊息channel的goroutine，一旦有訊息就發送給全部在線的user
func (this *Server) ListenMessage()  {
	//服務器端仍舊需要時刻監聽
	for {
		msg := <-this.Message
		this.mapLock.Lock()
		//將訊息發給所有用戶
		for _, u := range this.OnlineMap {
			u.C <- msg
		}
		this.mapLock.Unlock()
	}
}

//廣播訊息
func (this *Server) BroadCast(user *User, msg string)  {
	sendMessage := "[" + user.Addr + "]" + user.Name + ":" + msg
	this.Message <- sendMessage
}

func (this *Server) Handler(conn net.Conn)  {
	//當前連接的業務

	user := NewUser(conn, this)
        fmt.Printf("new connection %s:%s established...\n", user.Addr, user.Name)

	//用戶在線訊息封裝
	user.Online()

	//監聽用戶是否活躍的channel
	isLive := make(chan bool)

	//接受客戶端發送的訊息
	//每個客戶端都有一個go程處理客戶端讀的業務
	go func() {
		// 4K大小
		buf := make([]byte, 4096)
		for  {
			n, err := conn.Read(buf)
			//fmt.Println("n=", n, ",buf=", string(buf[0:n]))
			//n為成功讀取到的位元組數
			if n == 0{
				user.Offline()
				return
			}
			if err != nil && err != io.EOF{
				fmt.Println("Conn Read err:", err)
				return
			}

			//正常獲取訊息
			//用戶輸入訊息是以\n結尾的，需要去掉最后一個位元組
			msg := string(buf[:n-1])

			//fmt.Println("handler msg: ", msg)
			//接收到的訊息進行廣播
			user.DoMessage(msg)

			//用戶的任意訊息，代表用戶是一個活躍的用戶，激活管道
			isLive <- true
		}
	}()

	for {
		//當前handler阻塞監聽管道的訊息，一旦兩個管道有一個有值，就會執行select
		select {
			//這里一旦當isLive為True，那么就會進入select，執行完case<-isLive之后
			//會接著更新After管道，但是因為還沒到時間不會進入case<-After之后的陳述句
			case <- isLive:
				//	當前用戶被激活，
				//	這里為了重置定時器，把case<-isLive 放到了上邊

			//設定定時器，如果定時觸發，則強踢，如果發訊息，則重新激活定時器
			//After本身是一個channel，如果發生超時，那么該channel中就能讀取到資料
			case <- time.After(time.Second * 60):
				//進入case表示超時，重置定時器
				//將當前的User強制關閉

				//發出下線訊息
				user.SendMessage("you are forced offline")

				//銷毀管道資源
				close(user.C)

				//關閉用戶連接
				conn.Close()

				//可能是因為管道資源以及conn連接關閉之后
				//OnlineMap中值被回收之后，自動洗掉鍵值對？？？

				//退出handler
				return  // 或者runtime.Goexit()
		}
	}

}

//啟動服務器的介面
func (this *Server) Start()  {
	//socket listen
        fmt.Println("server is starting...")
	listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", this.Ip, this.Port))
	if err != nil{
		fmt.Println("listener err:", err)
		return
	}
	//為了防止遺忘關閉連接，加上defer保證在介面結束之后close
	//close listen socket
	defer listener.Close()

	//啟動監聽Message的goroutine
	go this.ListenMessage()

	//listener一直監聽連接
	for {
		//accept
		conn, err := listener.Accept()
		if err != nil{
			fmt.Println("listener accept err:", err)
			continue
		}
		//go程處理連接
		go this.Handler(conn)
	}
}

package main

import (
	"net"
	"strings"
)

type User struct {
	Name  string
	Addr  string
	C     chan string  // channel資料為string
	conn  net.Conn

	server *Server  // 當前用戶屬于的服務器
}

//創建用戶的介面
func NewUser(conn net.Conn, server *Server) *User {
	userAddr := conn.RemoteAddr().String()
	user := &User{
		Name: userAddr,
		Addr: userAddr,
		C:    make(chan string),
		conn: conn,
		server: server,
	}

	//每一個新用戶都系結一個go程監聽當前用戶的channel訊息
	go user.ListenMessage()

	return user
}

//處理用戶上線
func (this *User) Online()  {

	//用戶上線，將用戶加入onlinemap中
	this.server.mapLock.Lock()
	this.server.OnlineMap[this.Name] = this
	this.server.mapLock.Unlock()

	//廣播用戶上線的訊息
	this.server.BroadCast(this, "user is online")
}

//處理用戶下線
func (this *User) Offline()  {

	//用戶下線，將用戶從online map洗掉
	this.server.mapLock.Lock()
	delete(this.server.OnlineMap, this.Name)
	this.server.mapLock.Unlock()

	//廣播用戶下線
	this.server.BroadCast(this, "user offline")

}

//給當前user對應的客戶端發送訊息
func (this *User) SendMessage(msg string)  {
	this.conn.Write([]byte(msg))
}

//處理訊息
func (this *User) DoMessage(msg string)  {
	if msg == "who"{
		//查詢當前都有哪些用戶在線
		this.server.mapLock.Lock()
		for _, u := range this.server.OnlineMap{
			onlineMsg := "[" + u.Addr + "]" + u.Name + ":" + "online...\n"
			//只發送給當前發送查詢指令的用戶
			this.SendMessage(onlineMsg)
		}
		this.server.mapLock.Unlock()

	} else if len(msg) > 7 && msg[:7] == "rename|" {
		//新的用戶名
		newName := strings.Split(msg, "|")[1]
		//判斷新的用戶名是否存在
		_, ok := this.server.OnlineMap[newName]
		if ok{
			//服務器中已經存在該用戶名
			this.SendMessage(newName + "has existed...")
		} else {
			//修改OnlineMap
			this.server.mapLock.Lock()
			delete(this.server.OnlineMap, this.Name)
			this.server.OnlineMap[newName] = this
			this.server.mapLock.Unlock()

			this.Name = newName
			this.SendMessage("you have updated user name: " + this.Name + "\n")
		}
	} else if len(msg) > 4 && msg[:3] == "to|" {
		// 訊息格式 ”to|alex|hello“

		//1 獲取用戶名
		remoteName := strings.Split(msg, "|")[1]
		if remoteName == ""{
			//用戶名無效
			this.SendMessage("invalid, user \"to|alex|hello\"")
			return
		}

		//2 查詢物件
		remoteUser, ok := this.server.OnlineMap[remoteName]
		if !ok {
			//用戶不存在
			this.SendMessage("username not exist")
			return
		}

		//3 獲取通信訊息
		content := strings.Split(msg, "|")[2]
		if content == ""{
			this.SendMessage("invalid message")
			return
		}

		//4 發送訊息
		remoteUser.SendMessage(this.Name + " say: " + content)

	} else {
		this.server.BroadCast(this, msg)
	}
}

//監聽當前User channel的方法，一旦有訊息，就直接發送給對端客戶端
func (this *User) ListenMessage()  {
	for  {
		msg := <-this.C
		//連接寫回msg，轉換為二進制
		//fmt.Println("user listen msg:", msg)
		this.conn.Write([]byte(msg + "\n"))
	}
}

package main


func main()  {

	server := NewServer("127.0.0.1", 8888)
	server.Start()
}

package main

import (
	"flag"
	"fmt"
	"io"
	"net"
	"os"
)

type Client struct {
	ServerIp string
	ServerPort  int
	Name     string
	conn   net.Conn
	flag        int //判斷當前client的模式
}

func newClient(serverIp string, serverPort int) *Client {
	client := &Client{
		ServerIp: serverIp,
		ServerPort: serverPort,
		flag: 999,  // 設定flay默認值，否則flag默認為int整型
	}
	//創建鏈接
	conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", serverIp, serverPort))
	if err != nil {
		fmt.Println("net.Dial error:", err)
		return nil
	}
	client.conn = conn
	//回傳客戶端
	return client
}

//client選單欄的輸出，并獲取flag輸入
func (client *Client) menu() bool {
	var flag int

	fmt.Println("input 1 into public chat")
	fmt.Println("input 2 into private chat")
	fmt.Println("input 3 into rename")
	fmt.Println("input 0 into exit")

	fmt.Scanln(&flag)

	if flag >= 0 && flag <=3{
		client.flag = flag
		return true
	} else {
		fmt.Println("invalid input integer")
		return false
	}
}

// 監聽server回應的訊息，直接顯示到標準輸出
func (client *Client) DealResponse()  {
	io.Copy(os.Stdout, client.conn) // 永久阻塞監聽
	/*
	上面一句相當于如下for回圈一直從conn中讀取，然后輸出到終端
	//for {
	//	buf := make([]byte, 4096)
	//	client.conn.Read(buf)
	//	fmt.Println(string(buf))
	//}
	 */

}

//查詢在線用戶
func (client *Client) QueryUsers() {
	sendMsg := "who\n"  // 直接查詢
	_, err := client.conn.Write([]byte(sendMsg))
	if err != nil{
		fmt.Println("conn write err: ", err)
		return
	}
}

//私聊模式
func (client *Client) PrivateChat()  {
	var remoteName string
	var chatMsg string

	client.QueryUsers()
	fmt.Println("please chat username, exit for stop")
	fmt.Scanln(&remoteName)
	for remoteName != "exit"{
		fmt.Println("please input private chat content, exit for stop")
		fmt.Scanln(&chatMsg)
		for chatMsg != "exit"{
			//發送給服務器
			if len(chatMsg) != 0{
				//訊息不為空
				sendMsg := "to|" + remoteName + "|" + chatMsg + "\n\n"
				_, err := client.conn.Write([]byte(sendMsg))
				if err != nil{
					fmt.Println("conn write err:", err)
					break
				}
			}
			//寫入下一條訊息
			chatMsg = ""
			fmt.Println("please input private chat content, exit for stop")
			fmt.Scanln(&chatMsg)
		}
		//給一個用戶發送訊息之后，可能還會給其他用戶發送
		client.QueryUsers()
		fmt.Println("please chat username, exit for stop")
		fmt.Scanln(&remoteName)
	}

}

func (client *Client) PublicChat() {
	//公聊模式
	var chatMsg string

	fmt.Println("please input public chat content, exit for stop")
	fmt.Scanln(&chatMsg)

	for chatMsg != "exit"{
		//發送給服務器
		if len(chatMsg) != 0{
			sendMsg := chatMsg + "\n"
			_, err := client.conn.Write([]byte(sendMsg))
			if err != nil{
				fmt.Println("conn write err:", err)
				break
			}
		}

		//重新接受下一條訊息
		chatMsg = ""
		fmt.Println("please input public chat content, exit for stop")
		fmt.Scanln(&chatMsg)
	}
}

func (client *Client) UpdateName() bool {
	fmt.Println("please input username")
	//接收輸入的用戶名
	fmt.Scanln(&client.Name)

	sendMsg := "rename|" + client.Name + "\n"
	//按照格式寫入連接
	_, err := client.conn.Write([]byte(sendMsg))
	if err != nil{
		fmt.Println("conn write error:", err)
		return false
	}
	return true
}

func (client *Client) Run() {
	for client.flag != 0{
		for client.menu() != true {
		}

		//根據不同的模式處理不同的業務
		switch  client.flag {
			case 1:
				//公聊模式
				fmt.Println("under public chat mode...")
				client.PublicChat()
				break
			case 2:
				//私聊模式
				fmt.Println("under private chat mode...")
				client.PrivateChat()
				break
			case 3:
				//	改名
				fmt.Println("under rename mode...")
				client.UpdateName()
				break
			case 0:
				//退出
				fmt.Println("ready to exit")
				break
		}
	}
}

//嘗試從終端命令列決議IP和Port創建客戶端
var serverIp string
var serverPort int

//檔案的初始化函式
//命令的格式  ./client.exe -ip 127.0.0.1 -port 8888
func init() {
	//屬于初始化作業，一般放在init中
	flag.StringVar(&serverIp, "ip", "127.0.0.1", "set server ip(default:127.0.0.1)")
	flag.IntVar(&serverPort, "port", 8888, "set server port(default:8888)")
}

func main()  {
	//通過命令列決議
	flag.Parse()
	client := newClient(serverIp, serverPort)
	//client := newClient("127.0.0.1", 8888)
	if client == nil{
		fmt.Println("------- connect server error------")
		return
	}

	fmt.Println("-------- connect server success ------")

	//按理說啟動client.Run()方法之后，服務器回傳相應的處理結果，
	//主go程會阻塞在Run方法，如果使用主go程中的Run方法接受回傳訊息，就會變成串行執行
	//無法同一時刻滿足其他的業務，而run應該跟dealResponse應該是并行的
	//所以提供一個新的go程只處理server回應的資訊
	go client.DealResponse()

	// 啟動客戶端業務，主go程阻塞在Run方法
	fmt.Println("ready to process transactions......")
	client.Run()

}