尚硅谷大数据技术之电信客服

13) 创建类:IConverter

package com.atguigu.analysis.converter;

import com.atguigu.analysis.kv.base.BaseDimension;

import java.io.Closeable;

import java.io.IOException;

public interface IConverter {

    // 根据传入的dimension对象,获取数据库中对应该对象数据的id,如果不存在,则插入该数据再返回

    int getDimensionId(BaseDimension dimension) throws IOException;

}

14) 创建类:DimensionConverter

package com.atguigu.analysis.converter.impl;

import com.atguigu.analysis.converter.IConverter;

import com.atguigu.analysis.kv.base.BaseDimension;

import com.atguigu.analysis.kv.impl.ContactDimension;

import com.atguigu.analysis.kv.impl.DateDimension;

import com.atguigu.utils.JDBCCacheBean;

import com.atguigu.utils.JDBCUtil;

import com.atguigu.utils.LRUCache;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.io.IOException;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

/**

 * 维度对象转维度id类

 *

 * @author 尽际

 */

public class DimensionConverter implements IConverter {

    //日志记录类,注意导包的正确性

    private static final Logger logger = LoggerFactory.getLogger(DimensionConverter.class);

    //每个线程保留自己的Connection实例

    private ThreadLocal<Connection> threadLocalConnection = new ThreadLocal<>();

    //创建数据缓存队列

    private LRUCache<String, Integer> lruCache = new LRUCache(3000);

    public DimensionConverter() {

        // JVM虚拟机关闭时,尝试关闭数据库连接

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {

            logger.info(“stopping mysql connection…”);

            JDBCUtil.close(threadLocalConnection.get(), null, null);

            logger.info(“mysql connection is successfully closed”);

        }));

    }

    /**

     * 1、判断内存缓存中是否已经有该维度的id,如果存在则直接返回该id

     * 2、如果内存缓存中没有,则查询数据库中是否有该维度id,如果有,则查询出来,返回该id,并缓存到内存中。

     * 3、如果数据库中也没有该维度id,则直接插入一条新的维度信息,成功插入后,重新查询该维度,返回该id,并缓存到内存中。

     *

     * @param dimension

     * @return

     */

    @Override

    public int getDimensionId(BaseDimension dimension) throws IOException {

        //1、根据传入的维度对象取得该维度对象对应的cachekey

        String cacheKey = genCacheKey(dimension);

        //2、判断缓存中是否存在该cacheKey的缓存

        if (lruCache.containsKey(cacheKey)) {

            return lruCache.get(cacheKey);

        }

        //3、缓存中没有,查询数据库

        String[] sqls = null;

        if (dimension instanceof DateDimension) {

            // 时间维度表tb_dimension_date

            sqls = genDateDimensionSQL();

        } else if (dimension instanceof ContactDimension) {

            //联系人表tb_contacts

            sqls = genContactSQL();

        } else {

            //抛出Checked异常,提醒调用者可以自行处理。

            throw new IOException(“Cannot match the dimession, unknown dimension.”);

        }

        try {

            Connection conn = this.getConnection();

            int id = -1;

            synchronized (this) {

                id = execSQL(conn, sqls, dimension);

            }

            //将该id缓存到内存中

            lruCache.put(cacheKey, id);

            return id;

        } catch (SQLException e) {

            e.printStackTrace();

            throw new IOException(e);

        }

    }

    /**

     * LRUCACHE中缓存的键值对形式例如:<date_dimension20170820, 3>

     *

     * @param dimension

     * @return

     */

    private String genCacheKey(BaseDimension dimension) {

        StringBuilder sb = new StringBuilder();

        if (dimension instanceof DateDimension) {

            DateDimension dateDimension = (DateDimension) dimension;

            //拼装缓存id对应的key

            sb.append(“date_dimension”);

            sb.append(dateDimension.getYear()).append(dateDimension.getMonth()).append(dateDimension.getDay());

        } else if (dimension instanceof ContactDimension) {

            ContactDimension contactDimension = (ContactDimension) dimension;

            //拼装缓存id对应的key

            sb.append(“contact_dimension”);

            sb.append(contactDimension.getTelephone()).append(contactDimension.getName());

        }

        if (sb.length() <= 0) throw new RuntimeException(“Cannot create cachekey.” + dimension);

        return sb.toString();

    }

    /**

     * 生成时间维度的数据库查询语句和插入语句

     *

     * @return

     */

    private String[] genDateDimensionSQL() {

        String query = “SELECT `id` FROM `tb_dimension_date` WHERE `year` = ? AND `month` = ? AND `day` = ? order by `id`;”;

        String insert = “INSERT INTO `tb_dimension_date`(`year`, `month`, `day`) VALUES(?, ?, ?);”;

        return new String[]{query, insert};

    }

    /**

     * 生成联系人的数据库查询语句和插入语句

     *

     * @return

     */

    private String[] genContactSQL() {

        String query = “SELECT `id` FROM `tb_contacts` WHERE `telephone` = ? AND `name` = ? order by `id`;”;

        String insert = “INSERT INTO `tb_contacts`(`telephone`, `name`) VALUES(?, ?);”;

        return new String[]{query, insert};

    }

    /**

     * 尝试获取数据库连接对象:先从线程缓冲中获取,没有可用连接则创建。

     *

     * @return

     * @throws SQLException

     */

    private Connection getConnection() throws SQLException {

        Connection conn = null;

        synchronized (this) {

            conn = threadLocalConnection.get();

            if (conn == null || conn.isClosed() || conn.isValid(3)) {

                conn = JDBCCacheBean.getInstance();

            }

            threadLocalConnection.set(conn);

        }

        return conn;

    }

    /**

     * @param conn

     * @param sqls      第一个为查询语句,第二个为插入语句

     * @param dimension

     * @return

     */

    private int execSQL(Connection conn, String[] sqls, BaseDimension dimension) throws SQLException {

        PreparedStatement preparedStatement = null;

        ResultSet resultSet = null;

        try {

            //1、假设数据库中有该条数据

            //封装查询sql语句

            preparedStatement = conn.prepareStatement(sqls[0]);

            setArguments(preparedStatement, dimension);

            //执行查询

            resultSet = preparedStatement.executeQuery();

            if (resultSet.next()) {

                return resultSet.getInt(1);

            }

            //2、假设 数据库中没有该条数据

            //封装插入sql语句

            preparedStatement = conn.prepareStatement(sqls[1]);

            setArguments(preparedStatement, dimension);

            preparedStatement.executeUpdate();

            JDBCUtil.close(null, preparedStatement, resultSet);

            //重新获取id,调用自己即可。

            preparedStatement = conn.prepareStatement(sqls[0]);

            setArguments(preparedStatement, dimension);

            //执行查询

            resultSet = preparedStatement.executeQuery();

            if (resultSet.next()) {

                return resultSet.getInt(1);

            }

        } finally {

            JDBCUtil.close(null, preparedStatement, resultSet);

        }

        throw new RuntimeException(“Failed to get id”);

    }

    private void setArguments(PreparedStatement preparedStatement, BaseDimension dimension) throws SQLException {

        int i = 0;

        if (dimension instanceof DateDimension) {

            DateDimension dateDimension = (DateDimension) dimension;

            preparedStatement.setInt(++i, dateDimension.getYear());

            preparedStatement.setInt(++i, dateDimension.getMonth());

            preparedStatement.setInt(++i, dateDimension.getDay());

        } else if (dimension instanceof ContactDimension) {

            ContactDimension contactDimension = (ContactDimension) dimension;

            preparedStatement.setString(++i, contactDimension.getTelephone());

            preparedStatement.setString(++i, contactDimension.getName());

        }

    }

}


上一篇:
下一篇:
关于尚硅谷
教育理念
名师团队
学员心声
资源下载
视频下载
资料下载
工具下载
加入我们
招聘岗位
岗位介绍
招贤纳师
联系我们
全国统一咨询电话:010-56253825
地址:北京市昌平区宏福科技园综合楼6层(北京校区)

深圳市宝安区西部硅谷大厦B座C区一层(深圳校区)

上海市松江区谷阳北路166号大江商厦6层(上海校区)

武汉市东湖高新开发区东湖网谷(武汉校区)