gRPC 客户端不会消耗服务端的流量

亲爱的编程学习爱好者，如果你点开了这篇文章，说明你对《gRPC 客户端不会消耗服务端的流量》很感兴趣。本篇文章就来给大家详细解析一下，主要介绍一下，希望所有认真读完的童鞋们，都有实质性的提高。

我创建了一个 grpc 服务器端流服务（使用 rust），用于从 cluster a 上的 kafka 主题流式传输消息。每次外部客户端发起 websocket 连接时，cluster b 上的 go 客户端都会连接到 rpc 服务；消息通过此连接流式传输到外部客户端。

但是，go 客户端似乎无法可靠地从 rust 服务器消费（即使消息确实已提交）。即使 websocket 确实是从外部客户端读取信息；有时它似乎不从 grpc 流中消耗。我尝试阅读文档并找出可能导致此问题的任何资源泄漏，但无法解决问题。

我认为问题可能出在上下文处理中，或者围绕关闭连接（当流最初成功时，如果您关闭 websocket 并重新打开它，则消息将无法发送） p>

go客户端代码

type kafkachannel struct {
    client bridge.kafkastreamclient
}

func (kc *kafkachannel) consume(topic string) (<-chan []byte, func() error) {
    readchan := make(chan []byte)

    // context with cancellation to close the routine
    ctx, cancel := context.withcancel(context.background())
    stream, err := kc.client.consume(ctx, &bridge.consumerequest{
        topic: topic,
    })
    if err != nil {
        log.fatalf("error creating cosumer stream: %v", err)
    }

    // launch listener in new thread
    go func(reader *chan []byte, consumestream *bridge.kafkastream_consumeclient) {
        // recover from a panic (resources consumed)
        defer func() {
            if recover() != nil {
                log.println("consumer routine closed")
            }
        }()
        for {
            response, err := stream.recv()
            if err != nil {
                log.printf("error creating cosumer stream: %v", err)
                break
            }
            switch data := response.optionalcontent.(type) {
            case *bridge.kafkaresponse_content:
                *reader <- *&data.content
            default:
                break

            }
        }

    }(&readchan, &stream)
    // create a callback func that frees the resources
    closecallback := func() error {
        err := stream.closesend()
        close(readchan)
        cancel()
        return err
    }
    return readchan, closecallback
}

websocket

type chathandler struct {
    database       *db.queries
    client         *grpc.clientconn
    context        context.context
    sessionchannel chan []byte
}
func (handler *chathandler) getchatconnection(c *websocket.conn) {
    //initialisation...
    consumer, closeconsume := kchannel.consume(topic)

    for msg := range consumer {
        log.printf("received message from bridge: %s", string(msg))
        writemessagestart := time.now()

        if err = c.writemessage(1, msg); err != nil {
            log.printf("error writing message: %v", err)
            writemessageelapsed := time.since(writemessagestart)
            log.printf("write time elapsed error: %s", writemessageelapsed)
            if errors.is(err, syscall.epipe) {
                log.printf("sys error: %v", err)
                //continue
            }
            closeconsume()
            handler.close(c)
            return
        }
        writemessageelapsed := time.since(writemessagestart)
        log.printf("write time elapsed no error: %s", writemessageelapsed)
    }
}

rust 服务器端代码

为了完整性

async fn consume(
        &self,
        request: Request<ConsumeRequest>,
    ) -> Result<Response<Self::ConsumeStream>, Status> {
        let (tx, rx) = mpsc::unbounded_channel();
        info!("Initiated read-only stream");
        tokio::spawn(async move {
            let message = match Some(request.get_ref()) {
                Some(x) => x,
                None => return,
            };
            let topic = message.topic.clone();
            info!("Consuming on topic: {}", topic);
            let consumer = create_kafka_consumer(topic);
            loop {
                let result = consumer.stream().next().await;
                match result {
                    None => {
                        warn!("Received none-type from consumer stream");
                        continue;
                    }
                    Some(Err(e)) => {
                        error!("Error consuming from kafka broker: {:?}", e);
                        continue;
                    }
                    Some(Ok(message)) => {
                        let payload = match message.payload_view::<str>() {
                            None => {
                                warn!("Recived none-type when unpacking payload");
                                continue;
                            }
                            Some(Ok(s)) => {
                                info!("Received payload: {:?}", s);
                                s
                            }
                            Some(Err(e)) => {
                                error!("Error viewing payload contents: {}", e);
                                return;
                            }
                        };
                        info!("Received message from broker in read-only stream");
                        if payload.len() > 0 {
                            info!("Sending payload {:?}", payload);
                            match tx.send(Ok(KafkaResponse {
                                success: true,
                                optional_content: Some(
                                    bridge::kafka_response::OptionalContent::Content(
                                        (*payload).as_bytes().to_vec(),
                                    ),
                                ),
                            })) {
                                Ok(_) => info!("Successfully sent payload to client"),
                                Err(e) => {
                                    error!("GRPC error sending message to client {:?}", e);
                                    return;
                                }
                            }
                        } else {
                            warn!("No content detected in payload from broker");
                        }
                        match consumer.commit_message(&message, CommitMode::Async) {
                            Ok(_) => (),
                            Err(e) => {
                                error!("Error commiting a consumed message: {:?}", e);
                                return;
                            }
                        }
                    }
                }
            }
        });
        Ok(Response::new(Box::pin(
            tokio_stream::wrappers::UnboundedReceiverStream::new(rx),
        )))
    }

解决方案

问题是 go 客户端中的资源泄漏，在某种程度上，rust 服务器也是如此。

优雅地关闭 go 客户端流：

对于名为 consume 的流式 rpc

ctx, cancel := context.withcancel(context.background())
stream, err := handler.client.consume(ctx, &bridge.consumerequest{
    topic: topic,
})
closecallback := func() {
    stream.closesend()
    cancel()
    c.close() // where c := *websocket.conn
}
defer closecallback()

使用可取消上下文初始化流非常重要，以便运行时可以释放与多路复用 rpc 关联的资源。调用 cancel() 会关闭客户端的连接（当服务器尝试发送新消息时，这将在服务器上显示为错误）。

rust 在服务器端检测 client.closesend()。

在go客户端中调用stream.closesend()并不会通知服务器终止连接，它只是表明客户端将停止发送消息。在这种特殊情况下，我正在流式 rpc 连接中的异步线程中工作，因此为了在客户端与 websocket 断开连接时正常关闭连接，检测 clos​​esend 数据包并关闭 unboundedreceiver 通道非常重要命令关闭连接。

请注意，这样做的动机是服务器将继续尝试向“死亡”接收者发送数据包，因为否则它会认为连接仍然有效。

以下代码改编自this thread：

pub struct DropReceiver<T> {
    rx: mpsc::UnboundedReceiver<T>,
}

impl<T> Stream for DropReceiver<T> {
    type Item = T;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut self.rx).poll_recv(cx)
    }
}

// The Drop trait detects the `CloseSend`
impl<T> Drop for DropReceiver<T> {
    fn drop(&mut self) {
        info!("Connection has been droped");
        self.rx.close(); // Shutdown the receiver
    }
}

// Within the rpc

let (tx, rx) = mpsc::unbounded_channel();
let rx = DropReceiver { rx: rx };
tokio::spawn(async move {/** work */});
Ok(Response::new(Box::pin(rx))) // return this

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~