java - 使用 Spring JDBC 进行多线程

Question

我遇到了 Spring JDBC 和 SQL 数据库的连接问题。问题是，在第一次尝试时，我的方法创建了 n 个线程，它们查询数据库，但没有出现任何问题。如果我立即再次运行该方法，同样的事情——没问题。请注意，我不会在两次尝试之间重新启动应用程序。

当我在再次运行应用程序之前等待几分钟时会出现问题 - 所以我假设某处存在超时问题，或者线程正在被放弃。

要注意的是，当我使用单线程版本运行此方法时，它工作得非常好。所以我相信实际的 URL /user/pass/driver 设置没问题。我是多线程的新手，所以我认为我的实现中某处存在缺陷。

我正在使用 Spring JDBC 和 Apache Tomcat JNDI 连接池:

Java 多线程:

public List<Item> getSetPoints(List<Item> items) {
    ExecutorService executorService = Executors.newFixedThreadPool(10);
    for(Item item: items) {
        executorService.submit(new ProcessItem(item));
    }

    executorService.shutdown();
} 

class ProcessItem implements Runnable {
    private Item item;

    public ProcessItem(Item item) {
        this.item = item;
    }

    public void run() {
        Item newItem = piDAO.retrieveSetPoint(item);
    }
}

道:

@Component("PIDAO")
public class PIDAO {

    private NamedParameterJdbcTemplate jdbc;

    @Resource(name="pijdbc")
    public void setPiDataSource(DataSource jdbc) {
        this.jdbc = new NamedParameterJdbcTemplate(jdbc);
    }

    public Item retrieveSetPoint(Item item) {
        MapSqlParameterSource params = new MapSqlParameterSource();
        params.addValue("tag", item.getTagName());

        String sql = "SELECT TOP 1  time, value, status FROM piarchive.picomp2 WHERE tag = :tag AND status=0 AND questionable = false ORDER BY time DESC";
        try {
            return jdbc.queryForObject(sql, params, (rs, rowNum) -> {
                item.setPiDate(rs.getString("time"));
                item.setPiValue(rs.getString("value"));
                return item;
            });
        } catch (Exception e) {
           System.out.println(e);
        }
    }
}

Spring DAO 容器:

<jee:jndi-lookup jndi-name="jdbc/PI" id="pijdbc"
                 expected-type="javax.sql.DataSource">
</jee:jndi-lookup>

JNDI 配置:

 <Resource
    name="jdbc/PI"
    auth="Container"
    type="javax.sql.DataSource"
    maxTotal ="25"
    maxIdle="30"
    maxWaitMillis ="10000"
    driverClassName="com.osisoft.jdbc.Driver"
    url="**Valid URL**"
    username="**Valid Username**"
    password="**Valid Password**"
/>

发生错误时的堆栈跟踪:

org.springframework.jdbc.UncategorizedSQLException: PreparedStatementCallback; uncategorized SQLException for SQL [SELECT TOP 1  time, value, status FROM piarchive.picomp2 WHERE tag = ? AND status=0 AND questionable = false ORDER BY time DESC]; SQL state [null]; error code [0]; [Orb.Channel] The channel is not registered on server.; nested exception is java.sql.SQLException: [Orb.Channel] The channel is not registered on server.
at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:84)
at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:81)
at org.springframework.jdbc.support.AbstractFallbackSQLExceptionTranslator.translate(AbstractFallbackSQLExceptionTranslator.java:81)
at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:645)
at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:680)
at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:707)
at org.springframework.jdbc.core.JdbcTemplate.query(JdbcTemplate.java:757)
at org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate.queryForObject(NamedParameterJdbcTemplate.java:211)
at btv.app.dao.PIDAO.retrieveSetPoint(PIDAO.java:36)
at btv.app.service.PiService$ProcessItem.run(PiService.java:91)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.sql.SQLException: [Orb.Channel] The channel is not registered on server.
at com.osisoft.jdbc.PreparedStatementImpl.executeQuery(PreparedStatementImpl.java:167)
at org.apache.tomcat.dbcp.dbcp2.DelegatingPreparedStatement.executeQuery(DelegatingPreparedStatement.java:82)
at org.apache.tomcat.dbcp.dbcp2.DelegatingPreparedStatement.executeQuery(DelegatingPreparedStatement.java:82)
at org.springframework.jdbc.core.JdbcTemplate$1.doInPreparedStatement(JdbcTemplate.java:688)
at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:629)
... 11 more

错误:java.sql.SQLException:[Orb.Channel] channel 未在服务器上注册，对于我正在使用的特定驱动程序是唯一的。

经过一些研究，这实际上意味着与服务器的连接已断开 - 然而，查看 SQL 服务器上的日志表明它并没有在服务器端断开。

Answer 1

尝试在 retrieveSetPoint 方法中创建一个新的 NamedParameterJdbcTemplate 以查看这是否消除了您遇到的任何超时问题

@Component("PIDAO")
public class PIDAO {

    private DataSource jdbc;

    @Resource(name="pijdbc")
    public void setPiDataSource(DataSource jdbc) {
        this.jdbc = jdbc;
    }

    public Item retrieveSetPoint(Item item) {
        MapSqlParameterSource params = new MapSqlParameterSource();
        params.addValue("tag", item.getTagName());

        String sql = "SELECT TOP 1  time, value, status FROM piarchive.picomp2 WHERE tag = :tag AND status=0 AND questionable = false ORDER BY time DESC";
        try {
            return (new NamedParameterJdbcTemplate(jdbc)).queryForObject(sql, params, (rs, rowNum) -> {
                item.setPiDate(rs.getString("time"));
                item.setPiValue(rs.getString("value"));
                return item;
            });
        } catch (Exception e) {
           System.out.println(e);
        }
    }
}

或者，您可以重用 NamedParameterJdbcTemplates 并在它们超时时刷新它们；这可以从显式池中受益，例如

private final int poolSize = 10;

public Collection<Item> getSetPoints(List<Item> items) {
    ExecutorService executorService = Executors.newFixedThreadPool(poolSize);
    Queue<Item> queue = new ConcurrentLinkedQueue<>();
    queue.addAll(items);
    Collection<Item> output = new ConcurrentLinkedQueue<>();
    for(int i = 0; i < poolSize; i++) {
        executorService.submit(new ProcessItem(queue, output);
    }
    return output;
} 

class ProcessItem implements Runnable {
    private final Queue<Item> queue;
    private final Collection<Item> output;
    private NamedParameterJdbcTemplate jdbc;

    public ProcessItem(Queue<Item> queue, Collection<Item> output) {
        this.queue = queue;
        this.output = output;
        this.jdbc = piDAO.getNamedJdbcTemplate();
    }

    public void run() {
        Item item = null;
        while((item = queue.poll()) != null) {
            try {
                output.add(piDAO.retrieveSetPoint(item, jdbc));
            } catch(SQLException e) {
                this.jdbc = piDAO.getNamedJdbcTemplate();
                output.add(piDAO.retrieveSetPoint(item, jdbc));
            }
        }
    }
}

@Component("PIDAO")
public class PIDAO {

    private DataSource jdbc;

    @Resource(name="pijdbc")
    public void setPiDataSource(DataSource jdbc) {
        this.jdbc = jdbc;
    }

    public NamedParameterJdbcTemplate getNamedJdbcTemplate() {
        return new NamedParameterJdbcTemplate(jdbc);
    }

    public Item retrieveSetPoint(Item item, NamedParameterJdbcTemplate template) throws SQLException {
        MapSqlParameterSource params = new MapSqlParameterSource();
        params.addValue("tag", item.getTagName());

        String sql = "SELECT TOP 1  time, value, status FROM piarchive.picomp2 WHERE tag = :tag AND status=0 AND questionable = false ORDER BY time DESC";
        return template.queryForObject(sql, params, (rs, rowNum) -> {
            item.setPiDate(rs.getString("time"));
            item.setPiValue(rs.getString("value"));
            return item;
        });
    }
}