我已经开发了代码来做到这一点，也许它会有所帮助。在我的情况下，我正在流式传输图像并在ffplay中播放。我启动了3个线程，一个创建管道（在演示管道中执行'initializePipe'的管道），一个监听管道创建并启动fplay以及向管道写入图像的写入器线程。
demoPipes.cpp
#include "demoPipes.h"
using namespace std;
#define MAX_THREADS 100
//#define BUF_SIZE 255
void ErrorHandler(LPTSTR lpszFunction);
// Sample custom data structure for threads to use.
// This is passed by void pointer so it can be any data type
// that can be passed using a single void pointer (LPVOID).
/*typedef struct MyData {
int val1;
int val2;
} MYDATA, *PMYDATA;
*/
typedef struct
{
HANDLE pipe;
LPCSTR MEDIA_PIPE;
} Session;
int _tmain()
{
Session* s = (Session*)malloc(sizeof(Session));
s->pipe = NULL;
s->MEDIA_PIPE = "\\\\.\\pipe\\screenRec";
DWORD dwThreadIdArray[MAX_THREADS];
HANDLE hThreadArray[MAX_THREADS];
// Create MAX_THREADS worker threads.
for(int i=0; i<MAX_THREADS; i++)
{
//create pipe initializer thread
if(i==0){
hThreadArray[i] = CreateThread(
NULL, // default security attributes
0, // use default stack size
initializePipe, // thread function name
(LPVOID) s, // argument to thread function
0, // use default creation flags
&dwThreadIdArray[i]); // returns the thread identifier
}
//create writer thread
if(i>1){
PipeMetaData* pipe_data1 = fetch_file("D:\\images.png", s->pipe);
//PipeMetaData* pipe_data = createPipeMetaData(pipe_data1->file_vector,pipe_data1->byte_size);
//pipe_data1->pipe =s->pipe;
hThreadArray[i] = CreateThread(
NULL, // default security attributes
0, // use default stack size
send_media_to_recorder, // thread function name
pipe_data1, // argument to thread function
0, // use default creation flags
&dwThreadIdArray[i]); // returns the thread identifier
}
//create recorder thread
if(i==1){
hThreadArray[i] = CreateThread(
NULL, // default security attributes
0, // use default stack size
recorder_pipe_function, // thread function name
0, // argument to thread function
0, // use default creation flags
&dwThreadIdArray[i]); // returns the thread identifier
}
// Check the return value for success.
// If CreateThread fails, terminate execution.
// This will automatically clean up threads and memory.
if (dwThreadIdArray[i] == NULL)
{
ErrorHandler(TEXT("CreateThread"));
ExitProcess(3);
}
} // End of main thread creation loop.
//for close RTP
// Wait until all threads have terminated.
WaitForMultipleObjects(MAX_THREADS, hThreadArray, TRUE, INFINITE);
// Close all thread handles and free memory allocations.
CloseHandle(s->pipe);
for(int i=0; i<MAX_THREADS; i++)
{
CloseHandle(hThreadArray[i]);
}
getchar();
return 0;
}
void ErrorHandler(LPTSTR lpszFunction)
{
// Retrieve the system error message for the last-error code.
LPVOID lpMsgBuf;
LPVOID lpDisplayBuf;
DWORD dw = GetLastError();
FormatMessage(
FORMAT_MESSAGE_ALLOCATE_BUFFER |
FORMAT_MESSAGE_FROM_SYSTEM |
FORMAT_MESSAGE_IGNORE_INSERTS,
NULL,
dw,
MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
(LPTSTR) &lpMsgBuf,
0, NULL);
// Display the error message.
lpDisplayBuf = (LPVOID)LocalAlloc(LMEM_ZEROINIT,
(lstrlen((LPCTSTR) lpMsgBuf) + lstrlen((LPCTSTR) lpszFunction) + 40) * sizeof(TCHAR));
StringCchPrintf((LPTSTR)lpDisplayBuf,
LocalSize(lpDisplayBuf)/sizeof(TCHAR),
TEXT("%s failed with error %d: %s"),
lpszFunction, dw, lpMsgBuf);
MessageBox(NULL, (LPCTSTR) lpDisplayBuf, TEXT("Error"), MB_OK);
// Free error-handling buffer allocations.
LocalFree(lpMsgBuf);
LocalFree(lpDisplayBuf);
}
clientpipe.cpp
#include "client_pipe.h"
using namespace std;
DWORD WINAPI recorder_pipe_function(LPVOID lpParam){
const char* MEDIA_PIPE = "\\\\.\\pipe\\screenRec";
//LPCWSTR MEDIA_PIPE_wchar = L"\\\\.\\pipe\\screenRec";
//waits for server to initialize the media pipe.
//wait_for_media_pipe(MEDIA_PIPE,10);
WaitNamedPipeA(MEDIA_PIPE,1000);
start_recording_session(MEDIA_PIPE);
return 0;
}
void start_recording_session(const char* media_pipe){
//initialize recording command
char* command = (char*)malloc(sizeof(char)*100+1);
command[0] ='\0';
strcpy_s(command,sizeof(char)*100+1,"start ffplay -i ");
//strcpy_s(command,sizeof(char)*100+1,"ffmpeg -i ");
strcat_s(command,sizeof(char)*100+1, media_pipe);
//strcat_s(command,sizeof(char)*100+1, " -f matroska D:\\djhfifj.mkv");
std::cout << "Executing ffmpeg command\n";
int i=system (command);
std::cout << "The value returned was: " << i << std::endl;
}
ServerPipe.cpp
#include "server_pipe.h"
using namespace std;
typedef struct
{
HANDLE pipe;
LPCSTR MEDIA_PIPE;
} Session;
DWORD WINAPI initializePipe(LPVOID lpParam) {
printf("Creating an instance of a named pipe...");
Session* s = (Session*) lpParam;
LPCSTR MEDIA_PIPE = s->MEDIA_PIPE;
// Create a pipe to send data
s->pipe = CreateNamedPipeA(
MEDIA_PIPE, // name of the pipe
PIPE_ACCESS_OUTBOUND, // 1-way pipe -- send only
PIPE_TYPE_BYTE, // send data as a byte stream
1, // only allow 1 instance of this pipe
0, // no outbound buffer
0, // no inbound buffer
0, // use default wait time
NULL // use default security attributes
);
if (s->pipe == NULL ||s->pipe == INVALID_HANDLE_VALUE) {
wcout << "Failed to create outbound pipe instance.";
// look up error code here using GetLastError()
return NULL;
}
wcout << "Waiting for a client to connect to the pipe..." << endl;
// This call blocks until a client process connects to the pipe
BOOL result = ConnectNamedPipe(s->pipe, NULL);
if (!result) {
wcout << "Failed to make connection on named pipe." << endl;
// look up error code here using GetLastError()
CloseHandle(s->pipe); // close the pipe
system("pause");
return NULL;
}
}
PipeMetaData* fetch_file(char* file_path,HANDLE pipe){
PipeMetaData* data = (PipeMetaData*) malloc(sizeof(PipeMetaData));
//opening file
ifstream infile;
infile.open(file_path,std::ios::binary);
infile.seekg(0,std::ios::end);
//get file byte size
data->byte_size = infile.tellg();
//fetch data to send
data->file_vector =(char*) malloc(sizeof(char)*data->byte_size);
infile.seekg(0,std::ios::beg);
infile.read(&(data->file_vector)[0],data->byte_size);
data->packet = &(data->file_vector)[0];
data->pipe = pipe;
wcout<<data->byte_size<<endl;
return data;
}
DWORD WINAPI send_media_to_recorder(LPVOID lpParam){
PipeMetaData* p = (PipeMetaData*) lpParam;
// This call blocks until a client process reads all the data
DWORD numBytesWritten = 0;
BOOL result = WriteFile(
p->pipe, // handle to our outbound pipe
p->packet,//&(p->file_vector)[0], // data to send
p->byte_size, // length of data to send (bytes)
&numBytesWritten, // will store actual amount of data sent
NULL // not using overlapped IO
);
if (result) {
wcout << "Number of bytes sent: " << numBytesWritten << endl;
} else {
wcout << "Failed to send data." << endl;
// look up error code here using GetLastError()
GetLastError();
}
free(p);
return 0;
}