import com.amazonaws.services.kinesis.model.CreateStreamRequest; //导入依赖的package包/类
public void createStream(int shardCount, String streamName)
{
    CreateStreamRequest createStreamRequest = new CreateStreamRequest();
    createStreamRequest.setStreamName(streamName);
    createStreamRequest.setShardCount(shardCount);

    amazonKinesisClient.createStream(createStreamRequest);
    try {
        while (checkStreamStatus(streamName).equals("ACTIVE") == false) {
            MILLISECONDS.sleep(1000);
        }
    }
    catch (Exception e) {
    }

    streamsCreated.add(streamName);
}
 

import com.amazonaws.services.kinesis.model.CreateStreamRequest; //导入依赖的package包/类
/**
 * Create the specified topic with the specified number of partitions
 */
public void createTopic(String topicName, int partitions) {
	LOGGER.info("Determining if  Kinesis topic: {} already exists...", topicName);
	try{
		final DescribeStreamRequest describeRequest = new DescribeStreamRequest();
		describeRequest.withStreamName(topicName);
		this.client.describeStream(describeRequest);
		
	}catch(ResourceNotFoundException rnf){
		LOGGER.info("Kinesis stream for topic: {} does not exist, creating now with shard count: {}",topicName, partitions);
		final CreateStreamRequest request = new CreateStreamRequest();
		request.withStreamName(topicName);
		request.withShardCount(partitions);
		this.client.createStream(request);
	    this.waitForStreamToBecomeAvailable(topicName, DEFAULT_WAIT_TIME_MINUTES);
		LOGGER.info("Create topic completed for topic: {}", topicName);
	}    
}
 

import com.amazonaws.services.kinesis.model.CreateStreamRequest; //导入依赖的package包/类
@Override
public CreateStreamResult createStream(CreateStreamRequest createStreamRequest) throws AmazonServiceException, AmazonClientException
{
    // Setup method to create a new stream:
    InternalStream stream = new InternalStream(createStreamRequest.getStreamName(), createStreamRequest.getShardCount(), true);
    this.streams.add(stream);
    return new CreateStreamResult();
}
 

import com.amazonaws.services.kinesis.model.CreateStreamRequest; //导入依赖的package包/类
@Override
public CreateStreamResult createStream(CreateStreamRequest createStreamRequest) {
  throw new RuntimeException("Not implemented");
}