java netty websocket测压请教

ppgjx

我的服务器是 2g4h 我使用测压工具,创建了500个客户端,每1秒发送一条消息,服务器把这条消息发给所有客户端, 然后不一会这个服务器的cpu就满了 然后java就没响应了 提示 java.lang.OutOfMemoryError: GC overhead limit exceeded
这种问题好像是gc回收不了 然后造成java假死 我这服务器才500个链接就不行了 这种咋办呢? 还是说无解 服务器只能承载这样的并发 但是我看cpu是满了 内存其实还有1g多



这是代码处理:

[Asm] 纯文本查看 复制代码

@Component
@Slf4j
public class TestChannelHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {


    private static StringRedisTemplate redisTemplate;

    private static PxUserService pxUserService;


    @Resource
    public void setStringRedisTemplate(StringRedisTemplate redisTemplate) {
        TestChannelHandler.redisTemplate = redisTemplate;
    }



    @Resource
    public void setPxUserService(PxUserService pxUserService) {
        TestChannelHandler.pxUserService = pxUserService;
    }


    //用户id=>channel示例
    //可以通过用户的唯一标识保存用户的channel
    //这样就可以发送给指定的用户
    public static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);


    //客户端断开链接
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端断开连接 --- " + ctx.channel().id() );
        clients.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("channelRead");
        super.channelRead(ctx, msg);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        //获取客户端消息
        String context = msg.text();
        log.info(ctx.channel().id().asShortText() + " 测试数据 ---" + context + " -- 当前客户数量 --- " + clients.size());
 
        pushAllUser( "用户: " + ctx.channel().id().asShortText() + " 消息: " + context + " 当前客户数量: " + clients.size());
    }

    /**
     * 推送所有
     * [url=home.php?mod=space&uid=952169]@Param[/url] text
     */
    public void pushAllUser(String text){
        clients.writeAndFlush(new TextWebSocketFrame(text));
    }


    /**
     * 推送所有除了自己
     * @param text
     * @param ctx
     */
    public void pushOtherUser(String text,ChannelHandlerContext ctx){
        clients.forEach(ch ->{
            if(ctx.channel() != ch){
                ch.writeAndFlush(new TextWebSocketFrame(text));
            }
        });
    }

    //客户端建立连接
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws  Exception{

        System.out.println("客户端建立连接 --- " + ctx.channel().id() );
        //添加channels
        clients.add(ctx.channel());
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("异常 --- " + ctx.channel().id());
        ctx.channel().close();
        clients.remove(ctx.channel());
    }

Goldrepo

cpu是满了是因为频繁的gc操作，看你最后报java.lang.OutOfMemoryError: GC overhead limit exceeded导致异常，看下怎么优化jvm，减少gc的频率后再试试

xman55555

是 这个么 ?  https://github.com/YeautyYE/netty-websocket-spring-boot-starter

xman55555

[Java] 纯文本查看 复制代码

package xman.demo.utils;

import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.timeout.IdleStateEvent;
import org.springframework.util.MultiValueMap;
import org.yeauty.annotation.*;
import org.yeauty.pojo.Session;

import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

@ServerEndpoint(path = "/ws/{arg}",port = "8088")
public class MyWebSocket {

    //用户id=>channel示例
    //可以通过用户的唯一标识保存用户的channel
    //这样就可以发送给指定的用户
    public static Set<Session> clients = new HashSet<>();

    /**
     * 当有新的连接进入时，对该方法进行回调 注入参数的类型:Session、HttpHeaders...
     * [url=home.php?mod=space&uid=952169]@Param[/url] session
     * @param headers
     * @param req
     * @param reqMap
     * @param arg
     * @param pathMap
     */
    @BeforeHandshake
    public void handshake(Session session, HttpHeaders headers, @RequestParam String req, @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap){
        session.setSubprotocols("stomp");
        if (!"ok".equals(req)){
            System.out.println("Authentication failed!");
            session.close();
        }
    }
    
    @OnOpen
    public void onOpen(Session session, HttpHeaders headers, @RequestParam String req, @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap){
        System.out.println("new connection");
        System.out.println(req);
        clients.add(session);
    }

    @OnClose
    public void onClose(Session session) throws IOException {
       System.out.println("one connection closed");
        clients.remove(session);
    }

    @OnError
    public void onError(Session session, Throwable throwable) {
        throwable.printStackTrace();
    }

    /**
     * 当接收到字符串消息时，对该方法进行回调 注入参数的类型:Session、String
     * @param session
     * @param message
     */
    @OnMessage
    public void onMessage(Session session, String message) {
        clients.forEach(ch -> {
//            System.out.println(ch.id().asLongText());
            //自己不发送
            if(!session.channel().id().asLongText().equals(ch.channel().id().asLongText())){
                ch.sendText(message);
            }else {
//                ch.sendText("自己");
            }
        });
    }

    /**
     * 当接收到二进制消息时，对该方法进行回调 注入参数的类型:Session、byte[]
     * @param session
     * @param bytes
     */
    @OnBinary
    public void onBinary(Session session, byte[] bytes) {
        for (byte b : bytes) {
            System.out.println(b);
        }
        session.sendBinary(bytes); 
    }

    /**
     * 当接收到Netty的事件时，对该方法进行回调 注入参数的类型:Session、Object
     * @param session
     * @param evt
     */
    @OnEvent
    public void onEvent(Session session, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            switch (idleStateEvent.state()) {
                case READER_IDLE:
                    System.out.println("read idle");
                    break;
                case WRITER_IDLE:
                    System.out.println("write idle");
                    break;
                case ALL_IDLE:
                    System.out.println("all idle");
                    break;
                default:
                    break;
            }
        }
    }

}

随便写的,看是不是一样?感觉别人封装好的 应该会好用些

参考 : 调用  ws://127.0.0.1:8088/ws/xman2?req=ok