英文:
Trouble with Go Routines and its concurrency
问题
TL;DR 接受和连接两组不同的连接。希望能够同时使用RPC进行处理。
我正在尝试在我的计算机上创建一个半分布式系统。我目前正在处理的代码部分类似于代理节点,换句话说,它始终在一个端口上接受客户端连接。它还在不同的端口上不断接受后端节点的连接。
我试图找到一种同时监听并处理两者的方法。我正在使用RPC,我尝试的方法是这样的:
主函数:
func main() {
...
rpc.Register(myInterface)
l, err := net.Listen("tcp", client_port)
if err != nil {...}
go handleClients(l)
node_l, err := net.Listen("tcp", node_port)
if err != nil{...}
go setUpIncomingNodes(node_l, chan)
for{// 从chan中添加节点到列表}
}
并发函数:
// 添加要连接的客户端
func handleClients(listener net.Listener) {
for {
conn, err := listener.Accept()
if err != nil{...}
// 启动一个阻塞的(在该线程上)RPC连接
go rpc.serveConn(conn)
}
}
func setUpIncomingNodes(node_listener net.Listener, incoming_nodes chan<- net.Conn) {
for {
conn, err := node_listener.Accept()
if err != nil{...}
incoming_nodes <- conn
}
}
问题在于,它在第二个节点出现之前不会提供第一个节点的服务。我无法理解为什么会这样。而且似乎一次只能发生一个连接,但我认为RPC在不同的端口上提供服务(因此不会阻塞)。非常感谢任何帮助。
我尝试按照这个教程进行操作,但我发现情况有很大不同,它还使用了不同版本的Go。我使用了两种类型的节点/连接,其中类型A需要通过代理来为类型B提供服务。
英文:
TL;DR accepting and connecting two separate SETS of connections. Want to do it with RPC concurrently.
I'm attempting to create semi distributed system on my computer. The piece of the code I'm working on right now is something like a broker node, in other words it is always accepting client connections from a port. Its also constantly accepting back end node connections from a different port.
I'm trying to find a way to concurrently listen for both, and serve both. I'm using RPC and the way I tried doing it is this:
Main:
func main(){
...
rpc.Register(myInterface)
l, err := net.Listen("tcp", client_port)
if err != nil {...}
go handleClients(l)
node_l, err := net.Listen("tcp", node_port)
if err != nil{...}
go setUpIncomingNodes(node_l, chan)
for{// add nodes to a list from chan}
}
Concurrent functions:
// Adds clients to be connected
func handleClients(listener net.Listener){
for{
conn, err :=listener.Accept()
if err != nil{...}
// Starts up a blocking (on that thread) RPC connection
go rpc.serveConn(conn)
}
}
func setUpIncomingNodes(node_listener net.Listener, incoming_nodes chan<- net.Conn){
for{
conn, err := node_listener.Accept()
if err != nil{...}
incoming_nodes <- conn
}
}
The issue is that it doesn't serve the first node, until the second comes along. I can't understand why. Also it seems that only one connection can happen at time, but I thought RPC served on a different port (thus not blocking it). Any help is super appreciated.
I tried following this tutorial but I found the circumstance too different, its also using a different version of Go. I use two types of nodes/connections where Type A needs to be served through the broker to Type B.
答案1
得分: 1
最后,我认为问题在于我试图将监听器传递给Go协程,从而使每个Go协程与主线程有依赖关系。最终的解决方案就是这么简单:
func main() {
...
node_ip_port := "127.0.0.1:9000"
client_ip_port := "127.0.0.1:1000"
nodeChan := make(chan net.Conn, 20)
go func(ip_port string, nodeChan chan<- net.Conn) {
l, err := net.Listen("tcp", node_ip)
if err != nil {
log.Fatal("Had an error connecting to node", err)
}
for {
conn, _ := l.Accept()
kvChan <- conn
}
}(node_ip_port, nodeChan)
go func(ip_port string) {
l, err := net.Listen("tcp", ip_port)
if err != nil {
log.Fatal("Had an error connecting to client", err)
}
for {
conn, _ := l.Accept()
go rpc.ServeConn(conn)
}
}(client_ip_port)
// 连接节点的列表
nodeList := list.New()
for {
node := <-nodeChan
nodeList.PushBack(node)
// 我的RPC函数使用此列表中的节点进行服务 :-)
}
}
以上是翻译好的内容,请确认是否满意。
英文:
In the end, I think the issue was that I was attempting to pass in listeners to the Go routine, thus having each go routine having a dependancy with the main thread. The working solution ended up being as simple as this:
func main(){
...
node_ip_port := "127.0.0.1:9000"
client_ip_port := "127.0.0.1:1000"
nodeChan := make(chan net.Conn, 20)
go func(ip_port string, nodeChan chan<- net.Conn) {
l, err := net.Listen("tcp", node_ip)
if err != nil {
log.Fatal("Had an error connecting to node", err)
}
for {
conn, _ := l.Accept()
kvChan <- conn
}
}(node_ip_port, nodeChan)
go func(ip_port string) {
l, err := net.Listen("tcp", ip_port)
if err != nil {
log.Fatal("Had an error connecting to client", err)
}
for {
conn, _ := l.Accept()
go rpc.ServeConn(conn)
}
}(client_ip_port)
// List of connected nodes
nodeList := list.New()
for {
node := <-nodeChan
nodeList.PushBack(node)
// My RPC functions use the nodes in this list to serve :-)
}
}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论