netty+spring data mongodb,多线程保存数据,导致数据库连接数达到最大值。请问如何解决?

helloAndroid2016 发布于 2017/11/03 17:58
阅读 487
收藏 1

最近公司有个项目,需求是采集设备数据。数据量在1s内采集100条数据并入库。做技术调研时,想用netty做并发服务器,接收设备发送的数据,并保存在MongoDB中。采用netty+spring data MongoDB。

  问题有以下几点:

    1、netty接收数据的性能不高,每分钟只能接收650条左右。请问怎么设置netty的Acceptor线程池大小和work线程池大小,可达到更高的性能?

    2、在程序运行一段时间后,入库处理的线程会报如下异常:com.mongodb.MongoSocketReadException: Prematurely reached end of stream。在网上查了资料,说是数据库连接数已占用完。请问:为什么spring管理的连接池会一直占用数据库连接?在程序运行过程中,通过数据库连接工具查看数据库连接数,也能看到连接数在不断地增加。本人怀疑是spring配置MongoDB的配置文件不对,但却是按照spring官网配置的。

  附上本人代码,请帮忙看看,代码是否有什么问题。请大家指正,谢谢!

  spring配置文件app-context.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mongo="http://www.springframework.org/schema/data/mongo"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/data/mongo
        http://www.springframework.org/schema/data/mongo/spring-mongo-2.0.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-3.0.xsd">
    <mongo:repositories base-package="sinux.repository" />
    <!-- 获取配置资源 -->
    <context:property-placeholder location="classpath:mongodb.properties" />

    <mongo:mongo-client id="mongo-client" replica-set="${mongo.replicaSet}">
        <mongo:client-options
                connections-per-host="${mongo.connectionsPerHost}"
                threads-allowed-to-block-for-connection-multiplier="${mongo.threadsAllowedToBlockForConnectionMultiplier}"
                connect-timeout="${mongo.connectTimeout}"
                max-wait-time="${mongo.maxWaitTime}"
                socket-keep-alive="${mongo.socketKeepAlive}"
                socket-timeout="${mongo.socketTimeout}"/>
    </mongo:mongo-client>
    <!-- 设置使用的数据库名-->
    <mongo:db-factory dbname="netty" mongo-ref="mongo-client"/>
    <!-- mongodb的模板 -->
    <bean id="mongoTemplate" class="org.springframework.data.mongodb.core.MongoTemplate">
        <constructor-arg name="mongoDbFactory" ref="mongoDbFactory"/>
    </bean>
</beans>

  数据库连接配置文件mongodb.properties:

#mongo.replicaSet=localhost:27017
mongo.replicaSet=192.168.3.245:27017
mongo.connectionsPerHost=10
mongo.threadsAllowedToBlockForConnectionMultiplier=10
mongo.connectTimeout=10000
mongo.maxWaitTime=10000
mongo.autoConnectRetry=true
mongo.socketKeepAlive=false
mongo.socketTimeout=0
mongo.slaveOk=true
mongo.writeNumber=1
mongo.writeTimeout=0
mongo.writeFsync=true

  netty服务器代码Server.java:

package sinux.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.net.InetAddress;

/**
 * Created by  on 2017/10/25.
 * 服务器对象
 */
public class Server {
    private int server_port;

    public Server(int port){
        server_port = port;
    }

    public Server(){
        server_port = 8080;
    }

    public void start() {
        System.out.println("TCP服务器正在启动......");
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        EventLoopGroup bossGroup = new NioEventLoopGroup(4);
        EventLoopGroup workGroup = new NioEventLoopGroup(100);

        serverBootstrap.group(bossGroup,workGroup).channel(NioServerSocketChannel.class)
        .option(ChannelOption.SO_BACKLOG,100).handler(new LoggingHandler(LogLevel.INFO))
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ApplicationContext context = new ClassPathXmlApplicationContext("classpath*:/app-context.xml");
                ChannelPipeline channelPipeline = socketChannel.pipeline();
                channelPipeline.addLast(new LoggingHandler());
                channelPipeline.addLast(new StringEncoder());
                channelPipeline.addLast(new StringDecoder());
                channelPipeline.addLast(new ServerHandler(context));
            }
        });
        try {
            InetAddress address = InetAddress.getLocalHost();
            /**返回 IP 地址字符串(以文本表现形式)*/
            String serverIp = address.getHostAddress();
//            System.out.println("服务器IP:" + serverIp);
            ChannelFuture channelFuture = serverBootstrap.bind(serverIp,server_port).sync();
            System.out.println("TCP服务器启动成功!");
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

  服务器Handel类ServerHandler.java:

package sinux.server;

import com.alibaba.fastjson.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.transaction.annotation.Transactional;
import sinux.entity.DeviceTestData;
import sinux.entity.User;
import sinux.repository.DeviceTestDataRepository;
import sinux.repository.UserRepository;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by  on 2017/10/25.
 * 服务器适配器对象
 */
public class ServerHandler extends ChannelInboundHandlerAdapter {
    private ApplicationContext context;
//    private static Long executeTime = 0L;
    private static Long requestSize = 0L;
    private static Long processSize = 0L;

    public ServerHandler(ApplicationContext context){
        this.context = context;
    }

    @Override
    @Transactional(readOnly = false)
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        super.channelRead(ctx, msg);
        long startTime = System.currentTimeMillis();
        InetSocketAddress inSocket = (InetSocketAddress) ctx.channel()
                .remoteAddress();
        String clientIP = inSocket.getAddress().getHostAddress();
//        System.out.println("客户端(IP:" + clientIP + ")请求数据:" + msg);
        long endTime = System.currentTimeMillis();
        final DeviceTestData deviceTestData = (DeviceTestData)JSON.parseObject(msg.toString(),DeviceTestData.class);
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(100);
        fixedThreadPool.execute(new Runnable() {
            public void run() {
                deviceTestData.setTime(new Date());
                DeviceTestDataRepository deviceTestDataRepository = (DeviceTestDataRepository)context.getBean("deviceTestDataRepository");
                deviceTestDataRepository.save(deviceTestData);
                synchronized (processSize){
                    processSize ++;
                    System.out.println("处理了" + processSize + "个请求。");
                }
            }
        });
        synchronized (requestSize){
            requestSize ++;
            System.out.println("接收了" + requestSize + "个请求。");
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        super.channelReadComplete(ctx);
        ctx.flush();
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        cause.printStackTrace();
        ctx.close();
    }
}

  repository层DeviceTestDataRepository.java:

复制代码

package sinux.repository;

import org.springframework.data.mongodb.repository.MongoRepository;
import org.springframework.data.repository.CrudRepository;
import org.springframework.stereotype.Repository;
import sinux.entity.DeviceTestData;
import sinux.entity.User;

/**
 * Created by 黄利超 on 2017/10/31.
 */
@Repository
public interface DeviceTestDataRepository extends MongoRepository<DeviceTestData,String> {
}

复制代码

  请遇到过这种问题的大神说说解决方案,也恳请大家帮忙看看代码有什么不妥之处,欢迎大家指正!再次感谢大家!!!

加载中
0
h
helloAndroid2016

这个问题解决了!原因是启动spring容器的代码放错地方了。

见红框处,本人猜测:netty在接收请求时,都会执行这段代码,所以才导致netty接收数据慢,并会导致数据库连接不断增多。

返回顶部
顶部