在风暴螺栓中缓存

Caching in storm bolts

我正在尝试在 storm bolt 中缓存一些数据,但不确定这样做是否正确。在下面 class 中,员工 ID 和员工姓名被缓存到哈希映射中。为此,已对 Employee table 和 select 所有员工进行数据库调用,并在 prepare 方法中填充哈希映射(这是初始化映射的正确位置吗?)。

经过一些日志记录后发现(同时 运行 风暴拓扑),拓扑正在建立多个数据库连接并多次初始化地图。当然我想避免这种情况,这就是为什么我想缓存结果,这样它就不会每次都进入数据库。请帮忙?

public class TestBolt extends BaseRichBolt {
    private static final long serialVersionUID = 2946379346389650348L;
    private OutputCollector collector;
    private Map<String, String> employeeIdToNameMap;
    private static final Logger LOG = Logger.getLogger(TestBolt.class);

    @Override
    public void execute(Tuple tuple) {    
        String employeeId = tuple.getStringByField("employeeId");
        String employeeName = employeeIdToNameMap.get(employeeId);

        collector.emit(tuple, new Values(employeeId, employeeName));
        collector.ack(tuple);
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        // TODO Auto-generated method stub
        this.collector = collector;
        try {
            employeeIdToNameMap = createEmployeIdToNameMap();
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(/*some fields*/));

    }

    private Map<String, String> createEmployeIdToNameMap() throws SQLException {
        final Map<String, String> employeeIdToNameMap = new HashMap<>();
        final DatabaseManager dbm = new PostgresManager();
        final String query = "select id, name from employee;";
        final Connection conn = dbm.createDefaultConnection();
        final ResultSet result = dbm.executeSelectQuery(conn, query);
        while(result.next()) {
            String employeId = result.getString("id");
            String name = result.getString("name");
            employeeIdToNameMap.put(employeId, name);
        }
        conn.close();
        return employeeIdToNameMap;
    }       
}

解决方案 我创建了同步地图,它对我来说工作正常

private static Map<String, String> employeeIdToNameMap = Collections
            .synchronizedMap(new HashMap<String, String>());

因为你有多个bolt任务,你可以将employeeIdToNameMap标记为static和volatile。像这样在 prepare 中初始化地图 -

try {
 synchronized(TestBolt.class) {
    if (null == employeeIdToNameMap) {
    employeeIdToNameMap = createEmployeIdToNameMap();
    }
  }  
} catch (SQLException e) {
 ...
}