在风暴螺栓中缓存
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) {
...
}
我正在尝试在 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) {
...
}