database - 当每个 bolt 从同一点获取数据时，如何一个接一个地执行 bolt ？

Question

我正在从 spout 中获取数据。每个 bolt 都会将映射字段插入到我数据库中的不同表中。但是我的数据库表有约束。在我的测试表中，我有两个名为 user-details 和 my-details 的表，它们的约束允许首先填充用户表(首先应该插入)，然后仅插入我的详细信息表。当我运行拓扑时，仅插入用户表，因为当 bolt 对数据库执行插入查询时，它允许只有 psqlbolt 首先插入(由于约束)和 psqlbolt1 抛出异常说找不到用户 id。所以当我这样做时，我在 psqlbolt1 中保持 (1000)sleep 两个 bolt 正在工作。但是当我对许多 bolt 应用相同时( 12) 等待时间增加， bolt 执行失败，表示超出 bolt 等待时间。我如何才能先执行用户字段，然后只有 psql1 应该开始插入。

我的拓扑类

public class Topology {

ConnectionProvider cp;
protected static final String JDBC_CONF = "jdbc.conf";
protected static final String TABLE_NAME = "users";
protected static final String SELECT_QUERY = "select dept_name from department, user_department where department.dept_id = user_department.dept_id" +
        " and user_department.user_id = ?";
public static void main(String[] args) throws Exception{
    String argument = args[0];
    JdbcMapper jdbcMapper;
   TopologyBuilder builder = new TopologyBuilder();
Map map = Maps.newHashMap();
        map.put("dataSourceClassName", "org.postgresql.ds.PGSimpleDataSource");
    map.put("dataSource.url","jdbc:postgresql://localhost:5432/twitter_analysis?user=postgres");
ConnectionProvider cp = new MyConnectionProvider(map);

    jdbcMapper = new SimpleJdbcMapper(TABLE_NAME, cp);

    List<Column> schemaColumns = Lists.newArrayList(new Column("user_id", Types.INTEGER), new Column             ("user_name",Types.VARCHAR),new Column("create_date", Types.TIMESTAMP));



    JdbcMapper mapper = new SimpleJdbcMapper(schemaColumns);

        PsqlBolt userPersistanceBolt = new PsqlBolt(cp, mapper)
    .withInsertQuery("insert into user_details (id, user_name, created_timestamp) values (?,?,?)");

    builder.setSpout("myspout", new UserSpout(), 1);

    builder.setBolt("Psql_Bolt", userPersistanceBolt,1).shuffleGrouping("myspout");



    jdbcMapper = new SimpleJdbcMapper("My_details", cp);

    List<Column> schemaColumns1 = Lists.newArrayList(new Column("my_id", Types.INTEGER), new Column             ("my_name",Types.VARCHAR));



    JdbcMapper mapper1 = new SimpleJdbcMapper(schemaColumns1);

        PsqlBolt1 userPersistanceBolt1 = new PsqlBolt1(cp, mapper1)
    .withInsertQuery("insert into My_details (my_id, my_name) values (?,?)");

    //builder.setSpout("myspout", new UserSpout(), 1);

    builder.setBolt("Psql_Bolt1", userPersistanceBolt1,1).shuffleGrouping("myspout");
 Config conf = new Config();
     conf.put(JDBC_CONF, map);
     conf.setDebug(true);
         conf.setNumWorkers(3);


    if (argument.equalsIgnoreCase("runLocally")){
        System.out.println("Running topology locally...");
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("Twitter Test Storm-postgresql", conf, builder.createTopology());
    }
    else {
        System.out.println("Running topology on cluster...");
        StormSubmitter.submitTopology("Topology_psql", conf, builder.createTopology()); 
    }
}}

我的 bolt :psql1

public class PsqlBolt1 extends AbstractJdbcBolt {
private static final Logger LOG = Logger.getLogger(PsqlBolt1.class);

 private String tableName;
 private String insertQuery;
 private JdbcMapper jdbcMapper;

public PsqlBolt1(ConnectionProvider connectionProvider,  JdbcMapper jdbcMapper) {
    super(connectionProvider);
    this.jdbcMapper = jdbcMapper;
}
public PsqlBolt1 withInsertQuery(String insertQuery) {
    this.insertQuery = insertQuery;
System.out.println("query passsed.....");
    return this;
}
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
    super.prepare(map, topologyContext, collector);
    if(StringUtils.isBlank(tableName) && StringUtils.isBlank(insertQuery)) {
        throw new IllegalArgumentException("You must supply either a tableName or an insert Query.");
    }
}

@Override
public void execute(Tuple tuple) {
    try {
  Thread.sleep(1000);
        List<Column> columns = jdbcMapper.getColumns(tuple);
        List<List<Column>> columnLists = new ArrayList<List<Column>>();
        columnLists.add(columns);
        if(!StringUtils.isBlank(tableName)) {
            this.jdbcClient.insert(this.tableName, columnLists);
        } else {
            this.jdbcClient.executeInsertQuery(this.insertQuery, columnLists);
        }
        this.collector.ack(tuple);
    } catch (Exception e) {
        this.collector.reportError(e);
        this.collector.fail(tuple);
    }
}

@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

}}

psql bolt :

public class PsqlBolt extends AbstractJdbcBolt {
private static final Logger LOG = Logger.getLogger(PsqlBolt.class);
 private String tableName;
 private String insertQuery;
 private JdbcMapper jdbcMapper;

public PsqlBolt(ConnectionProvider connectionProvider,  JdbcMapper jdbcMapper) {
    super(connectionProvider);
    this.jdbcMapper = jdbcMapper;
}
public PsqlBolt withTableName(String tableName) {
    this.tableName = tableName;
    return this;
}



public PsqlBolt withInsertQuery(String insertQuery) {
    this.insertQuery = insertQuery;
System.out.println("query passsed.....");
    return this;
}
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) {
    super.prepare(map, topologyContext, collector);
    if(StringUtils.isBlank(tableName) && StringUtils.isBlank(insertQuery)) {
        throw new IllegalArgumentException("You must supply either a tableName or an insert Query.");
    }
}

@Override
public void execute(Tuple tuple) {
    try {

        List<Column> columns = jdbcMapper.getColumns(tuple);
        List<List<Column>> columnLists = new ArrayList<List<Column>>();
        columnLists.add(columns);
        if(!StringUtils.isBlank(tableName)) {
            this.jdbcClient.insert(this.tableName, columnLists);
        } else {
            this.jdbcClient.executeInsertQuery(this.insertQuery, columnLists);
        }
        this.collector.ack(tuple);
    } catch (Exception e) {
        this.collector.reportError(e);
        this.collector.fail(tuple);
    }
}

@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

}}

当我对许多 bolt 应用相同的方法时，我的拓扑颜色变为红色(等待状态)。

这是 bolt 等待时间。第一个 bolt 没有任何 sleep 。我在第二个 bolt 中保持 1 秒 sleep ，其余所有 bolt 都有 2 秒 sleep 。

如何替换那个 sleep 来执行我的工作，或者如果我增加 supervisor 的数量，问题会得到解决吗？

Answer 1

您可以重组您的拓扑结构，以便 spout 将消息 M 发送到 bolt 1。 bolt 1 可以对此消息执行一些操作，并且仅当操作成功时才​​将相同的消息转发到 bolt 2。这样， Action 之间就有了严格的顺序。