bm-site

Appearance

服务间通信超时问题怎么解决？ ​

服务间通信超时问题是微服务架构中常见的挑战之一，以下是一些有效的解决方案：

简答：服务间通信超时问题首先要做的是定位问题，分析问题，让后根据情况采取如下方案：

1，优化服务性能减少不必要的逻辑和数据库优化

2，优化网络配置调整超时时间设置和选择合适的通信协议

3，采用容错机制熔断器模式和重试机制

4，进行服务降级核心业务优先

5，优化服务架构异步通信和分批处理

6，监控和预警实时监控网络性能和预警机制

补充深入细节，附关键源码

一、优化服务性能 ​

  • 代码优化
    • 减少不必要的逻辑 ：对服务的业务逻辑进行审查，去除冗余的计算步骤和复杂的业务流程。例如，如果一个服务在处理请求时进行了大量的数据转换，但其中有些转换对于后续的业务处理并没有实际意义，就可以将其移除。
    • 优化算法 ：使用更高效的算法来处理数据。比如，在进行数据排序时，将时间复杂度较高的冒泡排序替换为快速排序，可以显著减少处理时间。
  • 数据库优化
    • 添加索引 ：为数据库中的查询字段添加索引，加快数据检索速度。例如，如果一个服务经常根据用户的手机号查询用户信息，那么在手机号字段上添加索引可以大大减少数据库查询时间。
    • 优化查询语句 ：避免使用复杂的嵌套查询和关联查询。例如，将多个表的关联查询拆分为多个简单的查询，然后在应用层进行数据整合。

二、优化网络配置 ​

  • 调整超时时间设置
    • 合理设置客户端超时时间 ：根据服务的实际响应时间和网络状况，设置合适的客户端超时时间。例如，对于一个通常响应时间为 200ms 的服务，可以将客户端超时时间设置为 500ms - 1000ms。

一、优化服务性能 ​

  1. 代码优化
    • 减少不必要的逻辑
      • 问题：服务中存在冗余的计算步骤。
      • 解决方案：审查代码，移除不必要的逻辑。
      • 示例代码：
java
/**
 * @Autho:TianMing
 * @Description： TODO
 */
public User getUserInfo(String userId) {
    User user = userDao.findById(userId);
    // 冗余的数据转换
    user.convertData();
    return user;
}

// 优化后的代码
public User getUserInfo(String userId) {
    return userDao.findById(userId);
}
- **优化算法**
    * **问题**：使用低效的排序算法。
    * **解决方案**：将冒泡排序替换为快速排序。
    * **示例代码**：
java
/**
 * @Autho:TianMing
 * @Description： TODO
 */
// 原代码（冒泡排序）
public static void bubbleSort(int[] arr) {
    for (int i = 0; i < arr.length - 1; i++) {
        for (int j = 0; j < arr.length - i - 1; j++) {
            if (arr[j] > arr[j + 1]) {
                int temp = arr[j];
                arr[j] = arr[j + 1];
                arr[j + 1] = temp;
            }
        }
    }
}

// 优化后的代码（快速排序）
public static void quickSort(int[] arr, int low, int high) {
    if (low < high) {
        int pivotIndex = partition(arr, low, high);
        quickSort(arr, low, pivotIndex - 1);
        quickSort(arr, pivotIndex + 1, high);
    }
}

private static int partition(int[] arr, int low, int high) {
    int pivot = arr[high];
    int i = low - 1;
    for (int j = low; j < high; j++) {
        if (arr[j] < pivot) {
            i++;
            int temp = arr[i];
            arr[i] = arr[j];
            arr[j] = temp;
        }
    }
    int temp = arr[i + 1];
    arr[i + 1] = arr[high];
    arr[high] = temp;
    return i + 1;
}
  1. 数据库优化
    • 添加索引
      • 问题：数据库查询性能差。
      • 解决方案：为常用查询字段添加索引。
      • 示例SQL：
sql
CREATE INDEX idx_user_phone ON users(phone);
- **优化查询语句**
    * **问题**：复杂的嵌套查询。
    * **解决方案**：拆分为简单查询。
    * **示例SQL**：
sql
-- 原复杂查询
SELECT u.* FROM users u WHERE u.id IN (SELECT user_id FROM orders WHERE status = 'completed');

-- 优化后的查询
SELECT u.* FROM users u JOIN (SELECT user_id FROM orders WHERE status = 'completed') o ON u.id = o.user_id;

二、优化网络配置和协议 ​

  1. 调整超时时间设置
    • 合理设置客户端超时时间
      • 问题：客户端超时时间设置不合理。
      • 解决方案：在客户端设置合理的超时时间。
      • 示例代码（RestTemplate）：
java
/**
 * @Autho:TianMing
 * @Description： TODO
 */
RestTemplate restTemplate = new RestTemplate();
restTemplate.setRequestFactory(new HttpComponentsClientHttpRequestFactory() {
    @Override
    protected HttpClient buildHttpClient() {
        return HttpClients.custom()
            .setDefaultRequestConfig(RequestConfig.custom()
                .setSocketTimeout(1000) // 设置超时时间为1秒
                .setConnectTimeout(1000)
                .build())
            .build();
    }
});
- **服务端设置响应时间限制**
    * **问题**：服务端响应时间过长。
    * **解决方案**：在服务端设置请求处理超时时间。
    * **示例代码（Tomcat）**：
xml
<!-- 在server.xml中配置 -->
<Connector port="8080" maxHttpHeaderSize="8192" maxThreads="150" minSpareThreads="25" maxSpareThreads="75" enableLookups="false" acceptCount="100" connectionTimeout="20000" disableUploadTimeout="true" />
  1. 选择合适的通信协议
    • 使用轻量级协议
      • 问题：使用重量级通信协议导致性能问题。
      • 解决方案：使用HTTP/2或gRPC。
      • 示例代码（gRPC）：
java
/**
 * @Autho:TianMing
 * @Description： 定义服务接口
 * @Date:2020/6/16 20:30
 */
public interface UserService {
    void getUser(UserRequest request, StreamObserver<UserResponse> responseObserver);
}

// 实现服务
public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {
    @Override
    public void getUser(UserRequest request, StreamObserver<UserResponse> responseObserver) {
        UserResponse response = UserResponse.newBuilder()
            .setId(request.getId())
            .setName("User Name")
            .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}

// 客户端调用
public class UserClient {
    public static void main(String[] args) {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
            .usePlaintext()
            .build();
        UserServiceGrpc.UserServicesStub stub = UserServiceGrpc.newStub(channel);
        UserRequest request = UserRequest.newBuilder().setId(1).build();
        stub.getUser(request, new StreamObserver<UserResponse>() {
            @Override
            public void onNext(UserResponse value) {
                System.out.println("User Name: " + value.getName());
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                System.out.println("Completed");
            }
        });
    }
}

3，能力足够的话可自定义消息协议

API工具包

java
/**
 * @Auth: TianMing
 * @Description： TODO
 * @Date:2020/6/16 20:31
 */
@Data
public class Header {
    //sessionId  , reqType , contextLen
    private long sessionId;
    private byte reqType;
    private int contextLen;
}
/**
 * @Auth: TianMing
 * @Description： TODO
 * @Date:2021/6/16 20:31
 */
@Data
public class MessageRecord {
    //Header  , Object  body
    private Header header;
    private Object body;
}
/**
 * @Auth: TianMing
 * @Description： TODO
 * @Date:2021/6/16 20:33
 */
public enum OpCode {
    REQ((byte)0),RES((byte)1),PING((byte)2),PONG((byte)3);
    private byte code;
    private OpCode(byte code) {
        this.code = code;
    }
    public byte code(){
        return  this.code;
    }
}

/**
 * @Auth: TianMing
 * @Description： TODO
 * @Date:2020/6/16 20:37
 */
//解码器
public class MessageRecordDecoder  extends ByteToMessageDecoder {
//    int length = 0;//拆包粘包解决方案3 定长读取
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        //长度不够，继续等待。超出长度交给下一个handler
//        if (byteBuf.readableBytes()>4){
//            if(length==0) length= byteBuf.readInt();
//            if(byteBuf.readableBytes()<length) return;
//        }
        System.out.println(">>>>>>>>>>>> 开始解码 >>>>>>>>>>>>");
        MessageRecord record = new MessageRecord();
        Header header = new Header();
        //按照消息协议的顺序
        header.setSessionId(byteBuf.readLong());//根据自定义的消息  long  byte
        header.setReqType(byteBuf.readByte());
        header.setContextLen(byteBuf.readInt());
        record.setHeader(header);
        if(header.getContextLen()>0) {
            byte[]contents = new byte[header.getContextLen()];
            byteBuf.readBytes(contents);
            ByteArrayInputStream bis = new ByteArrayInputStream(contents);
            ObjectInputStream ois = new ObjectInputStream(bis);
            record.setBody(ois.readObject());//反序列用的IO流
            System.out.println("反序列化的消息： "+record);
            list.add(record);
        }else{
            System.out.println("msg can't be null");
        }
    }
}

/**
 * @Auth: TianMing
 * @Description： TODO
 * @Date:2020/6/16 20:47
 */
public class MessageRecordEncoder extends MessageToByteEncoder<MessageRecord> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, MessageRecord messageRecord, ByteBuf byteBuf) throws Exception {
        System.out.println(">>>>>>>>>开始编码>>>>>>>>");
        Header header = messageRecord.getHeader();
        byteBuf.writeLong(header.getSessionId());
        byteBuf.writeByte(header.getReqType());
        Object  body = messageRecord.getBody();
        if (body!=null) {
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos);
            oos.writeObject(body);
            byte[]bytes = bos.toByteArray();
            byteBuf.writeInt(bytes.length);//解码器 读取 long 之后byte 之后就是这里的长度了
            byteBuf.writeBytes(bytes);
        }else{
            //消息体为空，则长度为0
            byteBuf.writeInt(0);
        }
    }
}

客户端：

java
/**
 *  @Auth:TianMing
 * @Description： TODO
 * @Date:2020/6/16 21:16
 */
public class ClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        MessageRecord record = (MessageRecord)msg;
        System.out.println("Client received msg : " + record.toString());
        super.channelRead(ctx, msg);
    }
}

/**
 * @Auth:TianMing
 * @Description： TODO
 * @Date:2020/6/16 21:09
 */
public class ProtocolClient {
    public static void main(String[] args) {
        EventLoopGroup worker = new NioEventLoopGroup();
        Bootstrap bootstrap = new  Bootstrap();
        bootstrap.group(worker).channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(
                                new LengthFieldBasedFrameDecoder(1024*1024,9,4,0,0))
                                .addLast(new MessageRecordEncoder())
                                .addLast(new MessageRecordDecoder())
                                .addLast(new ClientHandler());//处理服务端的消息。
                    }
                });
        try {
            ChannelFuture future=  bootstrap.connect(new InetSocketAddress("localhost",8888)).sync();
            System.out.println("服务端连接成功。。。。。。。。。。");
            Channel channel = future.channel();

            //开始传输消息
            // for (int i=0 ;i<100;i++) {
            MessageRecord record = new MessageRecord();
            Header header = new Header();
            header.setSessionId(1000);
            header.setReqType(OpCode.REQ.code());
            record.setHeader(header);
            String body = "this is my netty  protocol rpc  msg";
            record.setBody(body);
            channel.writeAndFlush(record);
            // }
            //同步等待
            future.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            //最终记得清理关闭
            worker.shutdownGracefully();
        }
    }
}

服务端：

java
/**
 * @Auth:TianMing
 * @Description： TODO
 * @Date:2020/6/16 21:02
 */
public class ServerHandler extends ChannelInboundHandlerAdapter {
    //channelRead

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        MessageRecord record = (MessageRecord)msg;
        System.out.println("server receive msg :" + record);
        record.setBody("server resp msg :"+record.getHeader().getSessionId());
        record.getHeader().setReqType(OpCode.RES.code());
        ctx.writeAndFlush(record);
    }
}

/**
 * @Auth:TianMing
 * @Description： TODO
 * @Date:2020/6/16 20:55
 */
public class ProtocolServer {
    public static void main(String[] args) {
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup work = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(boss,work)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //第一个handler是处理半包问题
                        socketChannel.pipeline().addLast(
                                new LengthFieldBasedFrameDecoder(1024*1024,9,4,0,0))
                                .addLast(new MessageRecordEncoder())
                                .addLast(new MessageRecordDecoder())
                                .addLast(new ServerHandler());//处理服务端的消息。
                    }
                });
        try {
           ChannelFuture channelFuture =  bootstrap.bind(8888).sync();
            System.out.println("我的服务已开启。。。。。。。。。");
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            work.shutdownGracefully();
            boss.shutdownGracefully();
        }
    }
}

三、采用容错机制 ​

  1. 熔断器模式
    • 实现原理
      • 问题：服务调用出现故障导致级联失败。
      • 解决方案：使用Hystrix实现熔断器。
      • 示例代码：
java
/**
 * @Autho:TianMing
 * @Description： 
 */
@HystrixCommand(fallbackMethod = "fallbackGetUser")
public User getUser(String userId) {
    return restTemplate.getForObject("http://user-service/users/" + userId, User.class);
}

public User fallbackGetUser(String userId) {
    return new User(userId, "Fallback User");
}
  1. 重试机制
    • 合理设置重试次数和间隔
      • 问题：网络抖动导致服务调用失败。
      • 解决方案：在客户端设置重试机制。
      • 示例代码（FeignClient）：
java
/**
 * @Autho:TianMing
 * @Description： 重试机制
 */
@FeignClient(name = "user-service", configuration = FeignConfig.class)
public interface UserClient {
    @GetMapping("/users/{userId}")
    User getUser(@PathVariable("userId") String userId);
}

@Configuration
public class FeignConfig {
    @Bean
    public Retryer feignRetryer() {
        return new Retryer.Default(1000, 3000, 3);
    }
}

四、进行服务降级 ​

  1. 核心业务优先
    • 剥离非核心功能
      • 问题：服务超时影响用户体验。
      • 解决方案：剥离非核心功能。
      • 示例代码：
java
/**
 * @Autho:TianMing
 * @Description： 服务降级
 */
public class ProductService {
    public ProductResponse getProductInfo(String productId) {
        Product product = productDao.findById(productId);
        // 剥离非核心功能，不加载评论信息
        return new ProductResponse(product);
    }
}
- **简化响应内容**
    * **问题**：响应内容过大导致传输时间长。
    * **解决方案**：返回简化的数据。
    * **示例代码**：
java
public class NewsService {
    public List<News> getNewsList() {
        // 返回简化的新闻列表
        return newsDao.findBriefNewsList();
    }
}

五、优化服务架构 ​

  1. 异步通信
    • 采用消息队列
      • 问题：服务间同步调用导致性能瓶颈。
      • 解决方案：使用RabbitMQ实现异步处理。
      • 示例代码：
java
/**
 * @Autho:TianMing
 * @Description： 生产者
 */
@Service
public class OrderProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendOrder(Order order) {
        rabbitTemplate.convertAndSend("order-exchange", "order.routing.key", order);
    }
}

// 消费者
@Service
public class OrderConsumer {
    @RabbitListener(queues = "order-queue")
    public void processOrder(Order order) {
        // 处理订单逻辑
        System.out.println("Processing order: " + order.getId());
    }
}
- **任务分批处理**
    * **问题**：批量处理任务耗时长。
    * **解决方案**：将任务拆分为多个小批次处理。
    * **示例代码**：
java
/**
 * @Autho:TianMing
 * @Description： 分批处理
 */
public class DataProcessor {
    public void processData(List<Data> dataList) {
        int batchSize = 100;
        for (int i = 0; i < dataList.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, dataList.size());
            List<Data> batch = dataList.subList(i, endIndex);
            processBatch(batch);
        }
    }

    private void processBatch(List<Data> batch) {
        // 处理一个批次的数据
        System.out.println("Processing batch of size: " + batch.size());
    }
}

六、监控和预警 ​

  1. 实时监控
    • 监控服务性能指标
      • 问题：无法实时掌握服务性能。
      • 解决方案：使用Prometheus和Grafana进行监控。
      • 示例代码（Spring Boot Actuator）：
java
/**
 * @Autho:TianMing
 * @Description：实时监控
 */
@SpringBootApplication
@EnableHypermediaSupport(type = HypermediaType.HAL)
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

// application.properties
management.endpoints.web.exposure.include=*
management.metrics.export.prometheus.enabled=true
- **监控网络状况**
    * **问题**：网络问题导致服务调用超时。
    * **解决方案**：使用网络监控工具。
    * **示例工具**：使用Wireshark或tcpdump监控网络流量。
  1. 预警机制
    • 设置合理的阈值
      • 问题：无法及时发现服务异常。
      • 解决方案：设置合理的预警阈值。
      • 示例代码（Prometheus Alertmanager）：
yaml
# prometheus.yml
alert.rules:
  - alert: HighResponseTime
    expr: http_request_duration_seconds{quantile="0.5"} > 0.5
    for: 1m
    labels:
      severity: warning
    annotations:
      summary: "High response time"
      description: "HTTP request duration exceeds 500ms for 1 minute."
- **及时响应预警**
    * **问题**：预警后无人处理。
    * **解决方案**：建立值班制度和自动化响应机制。
    * **示例工具**：使用PagerDuty或Opsgenie进行报警通知。

更新: 2025-04-22 17:27:29
原文: https://www.yuque.com/tulingzhouyu/db22bv/fgas3pl7edcy7p6s