多個程序操作同一張表導致超時的解決方案

當多個程序對同一張表進行操作時，容易出現超時現象，導致數據讀取或寫入失敗。以下是針對此問題的一些解決方法。

一、代碼優化

在進行大量數據讀寫操作時，代碼的效率顯得尤為重要。因此，進行代碼優化是一個有效的解決方法。常用的代碼優化方法有：

1、使用緩存，將頻繁讀寫的數據存入緩存。這樣可以減少數據庫的讀寫次數，提高程序的效率。


public class CacheHelper {
    private static Map cache = new HashMap();

    public static Object get(String key){
        return cache.get(key);
    }

    public static void set(String key, Object value){
        cache.put(key, value);
    }

    public static void remove(String key){
        cache.remove(key);
    }
}

//使用示例
Object obj = CacheHelper.get("key");
if (obj == null) {
    obj = getDataFromDatabase();
    CacheHelper.set("key", obj);
} 
return obj;

2、批量操作數據庫，將多條SQL語句合併為一條。這樣可以減少數據庫的連接次數，在處理大批量數據時效果更加明顯。


public void batchInsert(List data) {
    try {
        Connection conn = getConnection();
        conn.setAutoCommit(false);

        String sql = "insert into TABLE_NAME (col1, col2, col3, ... ) values(?, ?, ?, ...)";
        PreparedStatement ps = conn.prepareStatement(sql);

        for (int i=0; i<data.size(); (exception="" ...=""

二、分佈式鎖

在多個程序同時操作一張表時，避免重複寫入數據是一個重要的問題。使用分佈式鎖可以保證在任意時刻只有一個程序對數據進行寫操作。

1、使用Redis實現分佈式鎖：


public class RedisLock {
    private RedisTemplate redisTemplate;

    public RedisLock(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public boolean lock(String key) {
        ValueOperations valueOperations = redisTemplate.opsForValue();
        return valueOperations.setIfAbsent(key, "1");
    }

    public void unlock(String key) {
        redisTemplate.delete(key);
    }
}

//使用示例
RedisLock lock = new RedisLock(redisTemplate);
if (lock.lock("lock_key")) {
    //do something
    lock.unlock("lock_key");
} else {
    //wait or return directly
}

2、使用ZooKeeper實現分佈式鎖：


public class ZooKeeperLock {
    private ZooKeeper zooKeeper;

    public ZooKeeperLock(String zkUrl) throws IOException, KeeperException, InterruptedException {
        CountDownLatch connectedSignal = new CountDownLatch(1);
        zooKeeper = new ZooKeeper(zkUrl, 1000, new Watcher() {
            public void process(WatchedEvent we) {
                if (we.getState() == Event.KeeperState.SyncConnected) {
                    connectedSignal.countDown();
                }
            }
        });
        connectedSignal.await();
    }

    public boolean lock(String key) throws KeeperException, InterruptedException {
        String path = "/locks/" + key;
        String createdPath = zooKeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        List siblings = zooKeeper.getChildren("/locks", false);

        //重載排序方法(compareTo是從小到大排序)
        Collections.sort(siblings, new Comparator() {
            public int compare(String s1, String s2) {
                return s1.substring(s1.indexOf("-")+1).compareTo(s2.substring(s2.indexOf("-")+1));
            }
        });

        if (createdPath.equals("/locks/" + siblings.get(0))) {
            return true;
        } else {
            String prevSibling = siblings.get(siblings.indexOf(createdPath.substring("/locks/".length()))-1);
            final CountDownLatch lockSignal = new CountDownLatch(1);
            Stat stat = zooKeeper.exists("/locks/"+prevSibling, new Watcher() {
                public void process(WatchedEvent we) {
                    if (we.getType() == Event.EventType.NodeDeleted) {
                        lockSignal.countDown();
                    } else if (we.getType() == Event.EventType.None) {
                        if (we.getState() == Event.KeeperState.Expired) {
                            lockSignal.countDown();
                        }
                    }
                }
            });
            if (stat != null) {
                lockSignal.await();
            }
            return true;
        }
    }

    public void unlock(String key) throws KeeperException, InterruptedException {
        String path = "/locks/" + key;
        zooKeeper.delete(path, -1);
    }
}

//使用示例
ZooKeeperLock lock = new ZooKeeperLock("zk_url");
if (lock.lock("lock_key")) {
    //do something
    lock.unlock("lock_key");
} else {
    //wait or return directly
}

三、讀寫分離

當大量的讀操作與寫操作同時進行時，讀寫分離是一個有效的解決方法。將讀操作與寫操作分別交給不同的節點進行處理，可以有效減輕單個數據庫節點的負擔。

1、使用MySQL實現讀寫分離：


jdbc:mysql://master_server/database?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&rewriteBatchedStatements=true

jdbc:mysql:loadbalance://master_server,slave_server1,slave_server2/database?loadBalanceConnection=true&useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&rewriteBatchedStatements=true&readFromMasterWhenNoSlaves=true

2、使用MongoDB實現讀寫分離：


@Bean
public MongoClient mongoClient() {
    MongoClientURI uri = new MongoClientURI("mongodb://username:password@host1,host2,host3/?replicaSet=rs0");
    return new MongoClient(uri);
}

@Bean(name = "mongoTemplate")
public MongoTemplate mongoTemplate() throws Exception {
    MongoTemplate template = new MongoTemplate(mongoClient(), "database_name");
    return template;
}

四、分區表

當數據量過大時，將數據分成幾個小的表格是一種很好的解決方法。這樣可以避免單個表格過大，導致讀寫效率下降。

1、使用MySQL實現分區表：


CREATE TABLE sales (
     sale_id INT NOT NULL AUTO_INCREMENT,
     sale_date DATE NOT NULL,
     sale_amount DECIMAL(12,2) NOT NULL,
     PRIMARY KEY (sale_id,sale_date)
) 
PARTITION BY RANGE( YEAR(sale_date) ) (
     PARTITION p0 VALUES LESS THAN (2010),
     PARTITION p1 VALUES LESS THAN (2011),
     PARTITION p2 VALUES LESS THAN (2012),
     PARTITION p3 VALUES LESS THAN MAXVALUE
);

2、使用MongoDB實現分區表：


db.runCommand( { shardCollection: "db.collection", key: { "_id": "hashed" } } );

五、數據庫連接池

在進行大量數據讀寫操作時，數據庫連接池是一個非常重要的組件。它可以保證在並發高峰期，數據庫連接的利用率最大。

1、使用Druid實現數據庫連接池：


@Bean
public DataSource dataSource() {
    DruidDataSource dataSource = new DruidDataSource();
    dataSource.setDriverClassName("com.mysql.jdbc.Driver");
    dataSource.setUrl("jdbc:mysql://localhost:3306/test");
    dataSource.setUsername("root");
    dataSource.setPassword("root");
    dataSource.setMaxActive(20);
    dataSource.setInitialSize(1);
    dataSource.setMinIdle(1);
    dataSource.setMaxWait(60000);
    dataSource.setTimeBetweenEvictionRunsMillis(60000);
    dataSource.setMinEvictableIdleTimeMillis(300000);
    dataSource.setValidationQuery("SELECT 1 FROM DUAL");
    dataSource.setTestWhileIdle(true);
    dataSource.setTestOnBorrow(false);
    dataSource.setTestOnReturn(false);
    return dataSource;
}

@Bean
public JdbcTemplate jdbcTemplate() {
    return new JdbcTemplate(dataSource());
}

2、使用HikariCP實現數據庫連接池：


@Bean
public DataSource dataSource() {
    HikariConfig config = new HikariConfig();
    config.setDriverClassName("com.mysql.cj.jdbc.Driver");
    config.setJdbcUrl("jdbc:mysql://localhost:3306/test");
    config.setUsername("root");
    config.setPassword("root");
    config.setMaximumPoolSize(20);
    config.setMinimumIdle(1);
    config.setConnectionTestQuery("SELECT 1 FROM DUAL");
    config.setAutoCommit(false);
    config.addDataSourceProperty("cachePrepStmts", "true");
    config.addDataSourceProperty("prepStmtCacheSize", "250");
    config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
    config.addDataSourceProperty("useServerPrepStmts", "true");
    config.addDataSourceProperty("useLocalSessionState", "true");
    config.addDataSourceProperty("rewriteBatchedStatements", "true");
    config.addDataSourceProperty("cacheResultSetMetadata", "true");
    config.addDataSourceProperty("cacheServerConfiguration", "true");
    config.addDataSourceProperty("elideSetAutoCommits", "true");
    config.addDataSourceProperty("maintainTimeStats", "false");
    return new HikariDataSource(config);
}

@Bean
public JdbcTemplate jdbcTemplate() {
    return new JdbcTemplate(dataSource());
}

原創文章，作者：TYJVF，如若轉載，請註明出處：https://www.506064.com/zh-hk/n/373700.html